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

Question

Spark Dataframes具有一种withColumn可以一次添加一个新列的方法。要添加多个列，需要一个链withColumn。这是最佳做法吗？

我觉得使用mapPartitions具有更多优势。假设我有一个由3组成withColumn的链，然后有一个过滤器Row根据某些条件将s 去除。这是四个不同的操作（不过，我不确定其中是否有宽泛的转换）。但是如果我做得到，我可以一口气做完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)

第二部分，在和mapPartitions链上使用时要权衡些什么？withColumnfilter

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

Answer 1

这种方法是否有无法预料的问题？

多。最严重的影响是：

• 与普通DataFrame代码相比，内存占用量高出几倍，并且垃圾回收开销显着。
• 在执行上下文之间移动数据需要很高的序列化和反序列化成本。
• 在查询计划器中引入断点。
• 这样，toDF调用时进行模式推断的成本（如果提供了正确的模式，则可以避免），并且可能会重新执行所有先前的步骤。
• 等等...

其中的一些可以被避免udf和select/ withColumn，其他不能。

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

您mapPartitions不会删除Spark计划程序无法排除的任何操作，并且不提供任何优化。它的唯一优点是，它为昂贵的连接对象提供了一个不错的范围。

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

当您开始使用执行程序端Python逻辑时，您已经与Spark SQL产生分歧。不要紧，如果你使用udf，RDD或新加入的矢量UDF。最终，您应该根据代码的总体结构来做出决定-如果主要是直接对数据执行Python逻辑，那么最好还是坚持使用RDD或完全跳过Spark。

如果这只是逻辑的一小部分，并且不会引起严重的性能问题，请不要为此感到汗颜。