星火如何使用UDF与加入

Question

我想使用一个特定的UDF使用Spark

这里的计划：

我有一个table A（10万行）和table B（15百万行）

我想用table A和的table B 之一的UDF比较一个元素是有可能

下面是我的代码示例。在某些时候，我还需要说我UDF比较必须大于0,9更大：

DataFrame dfr = df 
       .select("name", "firstname", "adress1", "city1","compare(adress1,adress2)") 
       .join(dfa,df.col("adress1").equalTo(dfa.col("adress2")) 
         .and((df.col("city1").equalTo(dfa.col("city2")) 
           ...; 

这可能吗？

Answer 1

是的，可以。但是它会比正常的运营速度较慢，如星火将不能做谓语下推

例子：

val udf = udf((x : String, y : String) => { here compute similarity; }); 
val df3 = df1.join(df2, udf(df1.field1, df2.field1) > 0.9) 

例如：

val df1 = Seq (1, 2, 3, 4).toDF("x") 
val df2 = Seq(1, 3, 7, 11).toDF("q") 
val udf = org.apache.spark.sql.functions.udf((x : Int, q : Int) => { Math.abs(x - q); }); 
val df3 = df1.join(df2, udf(df1("x"), df2("q")) > 1) 

您也可以直接从用户返回布尔定义函数