如何从配置单元上下文中使用Pyspark调用用Java编写的配置单元UDF

Question

我在配置单元中使用getLastProcessedVal2 UDF从表中获取最新的分区。这个UDF是用java编写的。我想使用hive上下文使用pyspark中的相同UDF。

dfsql_sel_nxt_batch_id_ini=sqlContext.sql(''' select l4_xxxx_seee.**getLastProcessedVal2**("/data/l4/work/hive/l4__stge/proctl_stg","APP_AMLMKTE_L1","L1_AMLMKT_MDWE","TRE_EXTION","2.1")''')

错误：

执行错误。FunctionRegistry:无法加载UDF类：java.lang.ClassNotFoundException：

Answer 1

将pyspark shell启动为：

pyspark --jars /path/to.udf.jar <all-other-param>

或

提交带有--jars选项的pyspark作业为：

spark-submit --jars /path/to/udf.jar <all-other-param>

Answer 2

您可以使用SQLContext方法udf注册用户定义的函数，在那里您可以看到，您必须传递一个字符串作为第一个参数，它将表示使用SQL查询时udf的名称。

例如

sqlContext.udf().register("slen",
       (String arg1) -> arg1.length(),
       DataTypes.IntegerType);
sqlContext.sql("SELECT slen(name) FROM user").show();