Runner { void initializeUDFs(Config config) { if (!config.hasPath(UDFS_SECTION_CONFIG)) return; ConfigList udfList = config.getList(UDFS_SECTION_CONFIG); for (ConfigValue udfValue : udfList) { ConfigValueType udfValueType = udfValue.valueType(); if (!udfValueType.equals(ConfigValueType.OBJECT)) { throw new RuntimeException("UDF list must contain UDF objects"); } Config udfConfig = ((ConfigObject)udfValue).toConfig(); for (String path : Lists.newArrayList(UDFS_NAME, UDFS_CLASS)) { if (!udfConfig.hasPath(path)) { throw new RuntimeException("UDF entries must provide '" + path + "'"); } } String name = udfConfig.getString(UDFS_NAME); String className = udfConfig.getString(UDFS_CLASS); Contexts.getSparkSession().udf().registerJava(name, className, null); LOG.info("Registered Spark SQL UDF: " + name); } }  void run(Config config); static final String STEPS_SECTION_CONFIG; static final String TYPE_PROPERTY; static final String DATA_TYPE; static final String LOOP_TYPE; static final String DECISION_TYPE; static final String TASK_TYPE; static final String UDFS_SECTION_CONFIG; static final String UDFS_NAME; static final String UDFS_CLASS; static final String EVENT_HANDLERS_CONFIG; static final String CONFIG_LOADER_PROPERTY; static final String PIPELINE_THREADS_PROPERTY; }
@Test public void testValidUDFs() throws Exception { Contexts.closeSparkSession(); Config config = ConfigUtils.configFromResource("/udf/udf_valid.conf"); new Runner().initializeUDFs(config); Deriver deriver = ComponentFactory.create( Deriver.class, config.getConfig(Runner.STEPS_SECTION_CONFIG + ".runudf.deriver"), true); Dataset<Row> derived = deriver.derive(Maps.<String, Dataset<Row>>newHashMap()); assertEquals(RowFactory.create("hello", 1), derived.collectAsList().get(0)); }
@Test (expected = AnalysisException.class) public void testNoUDFs() throws Throwable { Contexts.closeSparkSession(); Config config = ConfigUtils.configFromResource("/udf/udf_none.conf"); new Runner().initializeUDFs(config); Deriver deriver = ComponentFactory.create( Deriver.class, config.getConfig("steps.runudf.deriver"), true); deriver.derive(Maps.<String, Dataset<Row>>newHashMap()); }