基于需要外部API调用的现有列创建新Spark数据帧列的最佳方法是什么？

Question

我有一个数据帧，我正在基于Python的Jupyter笔记本中使用。我想根据现有列的内容添加一个附加列，其中新列的内容派生自对原始列运行外部 API 调用。

我尝试的解决方案是使用基于Python的UDF。第一个单元格包含如下内容：

def analysis(old_column):

    new_column = myapi.analyze(text=old_column)
    return(new_column)

analysis_udf = udf(analysis)

第二个单元格是:

df2 = df1.withColumn("col2",analysis_udf('col1'))
df2.select('col2').show(n=5)

我的数据帧相对较大，大约有70000行，其中col1可以有100到10000个字符的文本。当我在单元2中运行上面的代码时，它实际上似乎运行得相当快（分钟），并转储了df2数据帧的5行。所以我以为我在做生意。然而，我的下一个手机有以下代码：

df2.cache()
df2.filter(col('col2').isNull()).count()

此代码的目的是缓存新数据帧的内容以缩短对 DF 的访问时间，然后计算数据帧中有多少条目具有 UDF 生成的空值。令人惊讶的是（对我来说）这花了很多时间来运行，最终提供了6的输出。我不清楚为什么第二个细胞运行得很快，而第三个细胞运行缓慢。但是，如果 df2.select（'col2'）.show（n=5） 调用会导致 UDF 在所有行上运行，并且该行的速度会很慢，然后访问数据帧新列的后续调用会很快。但事实并非如此，所以我当时认为缓存调用是导致UDF在所有行上运行的调用，因此现在任何后续调用都应该很快。所以添加了另一个单元格：

df2.show(n=5)

假设它会运行得很快，但同样，它花的时间比我预期的长得多，看起来也许UDF又开始运行了。(?)

我的问题是：

1. 哪些Spark api调用实际导致udf运行（或重新运行），以及如何构建调用以仅运行一次UDF，以便使用UDF的python函数输出的文本创建新列。
2. 我读到应该避免Python UDF，因为它们很慢（似乎是正确的），那么当我需要使用API调用来生成新列时，我有什么替代方案？

Answer 1

我认为df2.select（'col2'）. show（n=5）调用会导致UDF运行

这不是一个正确的假设。鉴于API的限制，Spark将评估尽可能少的数据。因为您使用Pythonudf，它将评估收集5行所需的最小分区数。

哪些火花api调用实际导致udf运行（或重新运行），以及如何构建调用以仅运行一次UDF，以便使用UDF的python函数输出的文本创建新列。

• 任何评估，如果数据不再缓存（从内存中逐出）。
• 可能是结果列的任何用法，除非 udf 被标记为非确定性。

我已经读到应该避免Python UDF，因为它们很慢（似乎是正确的），那么当我需要使用API调用来生成新列时，我有什么替代方案？

除非您想切换到Scala或RDD API，否则唯一的选择是pandas_udf，这稍微更有效，但只支持有限的类型子集。