火花:加入UDF或MAP功能

Question

我必须编写一个复杂的UDF，其中我必须用其他表进行连接，然后返回匹配的数量。实际用例要复杂得多，但是我在此简化了最小可再现的代码。这是UDF代码。

def predict_id(date,zip):
    filtered_ids = contest_savm.where((F.col('postal_code')==zip)  & (F.col('start_date')>=date))
    return filtered_ids.count()

当我使用以下代码定义UDF时，我会得到一个长列表的控制台错误：

predict_id_udf = F.udf(predict_id,types.IntegerType())

错误的最后一行是：

py4j.Py4JException: Method __getnewargs__([]) does not exist

我想知道什么是最好的方法。我也尝试过这样的map：

result_rdd = df.select("party_id").rdd
  .map(lambda x: predict_id(x[0],x[1]))
  .distinct()

这也导致了类似的最终错误。我想知道，如果有的话，我可以在UDF或MAP函数中对原始DataFrame的每一行进行连接。

Answer 1

我必须编写一个复杂的UDF，其中我必须与其他表进行连接，然后返回匹配项。

设计是不可能的。我想这样实现这样的效果，您必须使用高级DF/RDD操作员：

df.join(ontest_savm,
    (F.col('postal_code')==df["zip"])  & (F.col('start_date') >= df["date"])
).groupBy(*df.columns).count()