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

Question

Spark Dataframes有一个方法WithColumn每次添加一个新列。若要添加多列，需要一个With Column的链。这是做这件事的最佳实践吗？

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 DFs中使用可用的方法总是比推出自己的实现要好。如果我的论点错了，请让我知道。谢谢！欢迎所有的想法。

Answer 1

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

多重。最严重的影响是：

• 与普通dataframe代码相比，内存占用要高几倍，垃圾收集开销也很大。
• 在执行上下文之间移动数据所需的序列化和反序列化成本很高。
• 在查询计划程序中引入断点。
• 目前，TODF调用的模式推断成本（如果提供了正确的模式，可以避免）以及可能重新执行前面所有步骤的成本。
• 以此类推……

当您开始使用Executor端Python逻辑时，您已经与Spark SQL不同了。无论您使用UDF、RDD还是新添加的向量化UDF都无关紧要。最后，您应该根据代码的整体结构做出决定--如果它主要是直接在数据上执行的Python逻辑，那么使用rdd或完全跳过Spark可能会更好。

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