aaronshan / hive-third-functions Goto Github PK

    int leftArrayLength = leftArrayOI.getListLength(leftArray);
    int rightArrayLength = rightArrayOI.getListLength(rightArray);

    // Check if array is null or empty
    if (leftArray == null || rightArray == null || leftArrayLength < 0 || rightArrayLength < 0) {
        return null;
    }

    if (leftArrayLength == 0) {
        return leftArray;
    }
    if (rightArrayLength == 0) {
        return rightArray;
    }

    List<?> leftList = leftArrayOI.getList(leftArray);
    List<?> rightList = leftArrayOI.getList(rightArray);
    HashSet<?> result_set = Sets.newHashSet(leftList);
    result_set.retainAll(rightList);

    return new ArrayList(result_set);
}`

// register UDF
spark.sql("create temporary function id_card_province as 'cc.shanruifeng.functions.card.UDFChinaIdCardProvince'");
        	
// get file
Dataset<Row> rawdata = spark.read().csv("./src/main/resources/starM.csv");

// use UDF
rawdata.createOrReplaceTempView("starM");
Dataset<Row> udfModified = spark.sql("SELECT *, id_card_province(_c13) FROM starM");
udfModified.show();

org.apache.hadoop.hive.ql.metadata.HiveException: Unable to execute method public org.apache.hadoop.io.Text cc.shanruifeng.functions.card.UDFChinaIdCardProvince.evaluate(org.apache.hadoop.io.Text)  on object cc.shanruifeng.functions.card.UDFChinaIdCardProvince@5c622859 of class cc.shanruifeng.functions.card.UDFChinaIdCardProvince with arguments {652423184510291234:org.apache.hadoop.io.Text} of size 1
	at org.apache.hadoop.hive.ql.exec.FunctionRegistry.invoke(FunctionRegistry.java:981)
	at org.apache.spark.sql.hive.HiveSimpleUDF.eval(hiveUDFs.scala:91)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply_6$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
	at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:235)
	at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
	at org.apache.spark.scheduler.Task.run(Task.scala:108)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.reflect.InvocationTargetException
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.hadoop.hive.ql.exec.FunctionRegistry.invoke(FunctionRegistry.java:957)
	... 18 more
Caused by: java.lang.NullPointerException
	at org.apache.hadoop.io.Text.encode(Text.java:450)
	at org.apache.hadoop.io.Text.set(Text.java:198)
	at cc.shanruifeng.functions.card.UDFChinaIdCardProvince.evaluate(UDFChinaIdCardProvince.java:23)
	... 23 more

create table temp.test_explode (userid string, log string) partitioned by (day string) stored as orc;
insert into table temp.test_explode partition (day = '2017-11-01') values ('u1', '[{  "action": "a" },  {  
"action": "b"} ]');
insert into table temp.test_explode partition (day = '2017-11-02') values ('u2', '[{  "action": "a" , "arg1": "1"},  {  "action": "b", "arg2": "2"} ]');


create temporary function udf_json_array_extract as 'cc.shanruifeng.functions.json.UDFJsonArrayExtract';

create temporary function udf_json_array_extract_scalar as 'cc.shanruifeng.functions.json.UDFJsonExtractScalar';


select userid, udf_json_array_extract ( log, '$.action' ) from  temp.test_explode  where day = '2017-11-01';

-- result
u1	["\"a\"","\"b\""]

select userid, udf_json_array_extract_scalar ( log, '$.action' ) from  temp.test_explode  where day = '2017-11-01';

-- result
u1	NULL