PySpark：向 DataFrame 添加更多列的最佳实践答案

【问题标题】：PySpark: Best practice to add more columns to a DataFramePySpark：向 DataFrame 添加更多列的最佳实践
【发布时间】：2018-09-14 00:25:58
【问题描述】：

Spark Dataframes 有一个方法 withColumn 一次添加一个新列。要添加多个列，需要withColumns 链。这是执行此操作的最佳做法吗？

感觉使用mapPartitions更有优势。假设我有一个由三个withColumns 组成的链，然后一个过滤器根据特定条件删除Rows。这是四种不同的操作（不过，我不确定这些操作是否是广泛的转换）。但如果我做一个mapPartitions，我可以一口气做完。如果我有一个我希望每个 RDD 分区打开一次的数据库连接，这也会有所帮助。

我的问题分为两部分。

第一部分，这是我对 mapPartitions 的实现。这种方法是否有任何不可预见的问题？有没有更优雅的方法来做到这一点？

df2 = df.rdd.mapPartitions(add_new_cols).toDF()

def add_new_cols(rows):
    db = open_db_connection()
    new_rows = []
    new_row_1 = Row("existing_col_1", "existing_col_2", "new_col_1", "new_col_2")
    i = 0
    for each_row in rows:
        i += 1
        # conditionally omit rows
        if i % 3 == 0:
            continue
        db_result = db.get_some_result(each_row.existing_col_2)
        new_col_1 = ''.join([db_result, "_NEW"])
        new_col_2 = db_result
        new_f_row = new_row_1(each_row.existing_col_1, each_row.existing_col_2, new_col_1, new_col_2)
        new_rows.append(new_f_row)

    db.close()
    return iter(new_rows)

第二部分，在withColumn 和filter 链上使用mapPartitions 的权衡是什么？

我在某处读到，使用 Spark DF 的可用方法总是比推出自己的实现更好。如果我的论点是错误的，请告诉我。谢谢！欢迎所有想法。

【问题讨论】：

你能分享一个你无法解决的问题的例子吗？到目前为止，您的问题太宽泛，有点不清楚。
如果您担心的话，withColumns 链将不会连续执行。惰性火花将优化操作。
为了避免两次打开数据库连接，您可以返回一个列表，然后将输出拆分为列。像df = df.withColumn('list_output', myUDF()).select("*", col('list_output')[0].alias('new_col1'), col('list_output)[1].alias('new_col2')).drop("list_output") 这样的东西。转换为 rdd 并返回 DF 很慢，但我不是这方面的专家。
@void 看看this post。您也可以花哨并从您的 udf 返回 StructType()，然后使用 list_output.*
另见：How to assign the result of udf to multiple columns 和 How to add multiple columns using udf。

标签： apache-spark pyspark apache-spark-sql pyspark-sql

【解决方案1】：

这种方法是否存在任何不可预见的问题？

多个。最严重的影响是：

与普通的 DataFrame 代码相比，内存占用量高出几倍，而且垃圾回收开销也很大。
在执行上下文之间移动数据需要高成本的序列化和反序列化。
在查询计划器中引入断点。
照原样，toDF 调用的架构推断成本（如果提供正确的架构可以避免）以及可能重新执行所有前面的步骤。
等等……

其中一些可以通过udf 和select / withColumn 避免，其他则不能。

假设我有一个由三个 withColumns 组成的链，然后是一个过滤器来根据特定条件删除行。这是四种不同的操作（不过，我不确定这些操作是否是广泛的转换）。但是如果我做一个mapPartitions，我可以一次完成所有的事情

您的mapPartitions 没有删除任何操作，也没有提供任何优化，这是 Spark 规划器无法排除的。它唯一的优点是它为昂贵的连接对象提供了一个很好的范围。

我在某处读到，使用 Spark DF 的可用方法总是比推出自己的实现更好

当您开始使用执行器端 Python 逻辑时，您已经与 Spark SQL 不同了。不管你使用udf、RDD 还是新添加的矢量化udf。归根结底，您应该根据代码的整体结构做出决定——如果主要是直接在数据上执行的 Python 逻辑，最好坚持使用 RDD 或完全跳过 Spark。

如果它只是逻辑的一小部分，并且不会导致严重的性能问题，请不要为此烦恼。

【讨论】：

【解决方案2】：

使用df.withColumn() 是添加列的最佳方式。都是懒惰添加的

【讨论】：

我同意。不是所有的转换都被懒惰地评估了吗？甚至mapPartition 也应该如此。请浏览问题的 cmets。我提出了一个我必须跳起来的问题。
好的，所以我阅读了其他 cmets 以及您的要求。我对python语法不是很熟悉，所以在这里猜测它应该和scala一样。因此，在 Scala API 中，您传递给 add_new_cols 的 rows 实际上是 Iterator[Row]。为了在查询中使用这些行的内容，您正在具体化这些迭代器，这样做会耗尽这些迭代器。
我建议您不要在这种情况下使用withColumns，因为您最终会为每一行建立/使用一个/单独的连接。或者，使用 mapPartitions 并在每个分区级别上连接到外部数据库，其中连接来自某个单例池
我不太熟悉这里的问题陈述，但如果我要解决这个问题，我可能会使用 Spark API 将外部表作为 JDBC 源读取，在这个 DF 和我现有的一个，然后加入他们。在这个加入的 DF 上，我可以做所有的处理