apache-spark - 激发如何在联接中使用UDF

Question

我想使用特定的UDF和Spark
这是计划:

我有一个table A(一千万行)和一个table B(一千五百万行)

我想使用UDF来比较table A的一个元素和table B的一个元素
是否可以

这是我的代码示例。在某些时候，我还需要说我的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

是的你可以。但是，由于Spark无法进行谓词下推，因此它将比普通运算符要慢

例子:

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)

您也可以从用户定义函数直接返回 bool 值