转换蟒蛇 Lambda 函数而不将值返回到派斯帕克

Question

我在Python中有一个有效的lambda函数，它计算dataset1中的每个字符串和dataset2中的字符串之间的最高相似度。在迭代过程中，它将字符串、最佳匹配和相似性以及一些其他信息写入bigquery。没有返回值，因为该函数的目的是向bigquery数据集中插入一行。这个过程需要相当长的时间，这就是为什么我想使用Pyspark和Dataproc来加速这个过程。

将熊猫的数据帧转换成spark很容易。我在注册我的udf时遇到了麻烦，因为它没有返回值，而pyspark需要一个。此外，我不明白如何将python中的‘apply’函数映射到pyspark变体。所以基本上我的问题是如何转换下面的python代码来处理spark数据帧。

以下代码适用于常规 Python 环境：

def embargomatch(name, code, embargo_names):
     find best match 
     insert best match and additional information to bigquery

customer_names.apply(lambda x: embargoMatch(x['name'], x['customer_code'],embargo_names),axis=1)

因为pyspark需要一个返回类型，所以我在udf中添加了“return 1 ”,并尝试了以下方法:

customer_names = spark.createDataFrame(customer_names)

from pyspark.sql.types import IntegerType
embargo_match_udf = udf(lambda x: embargoMatch(x['name'], x['customer_code'],embargo_names), IntegerType())

现在我一直在尝试应用select函数，因为我不知道要给出什么参数。

Answer 1

我怀疑您对如何将多列传递给udf感到困惑——这是这个问题的一个很好的答案：Pyspark：在UDF中传递多列。

与其基于包装函数的 lambda 创建 udf，不如考虑通过直接基于禁运匹配创建 udf 来简化。

embargo_names = ...

# The parameters here are the columns passed into the udf
def embargomatch(name, customer_code):
    pass
embargo_match_udf = udf(embargomatch, IntegerType())
customer_names.select(embargo_match_udf(array('name', 'customer_code')).alias('column_name'))

话虽如此，怀疑您的udf不会返回任何内容 - 我通常将udfs视为向数据帧添加列的一种方式，但不会产生副作用。如果要将记录插入到 bigquery 中，请考虑执行如下操作：

customer_names.select('column_name').write.parquet('gs://some/path')
os.system("bq load --source_format=PARQUET [DATASET].[TABLE] gs://some/path")