Spark数据框中多列计算的优化方式？

Question

解决以下用例的优化或最佳性能方法是什么

考虑一个包含 100 万行和 100 列的数据框，我们对其中的 1 列感兴趣 - 消息。我需要根据消息中匹配关键字的存在条件构建 3 个新列。

• 消息：堆栈溢出对代码开发的贡献是 一天比一天增加
• flag1 关键字：堆栈、松弛
• flag2 关键字：twitter、facebook、whatsapp
• flag3 关键字：流、运行、增加

预期输出：(message,flag1,flag2,flag3) 堆栈溢出对代码开发的贡献日益增加,1,0,0

方法 1

 val tempDF = df.withColumn("flag1",computeFlag(col("message"))).withColumn("flag2",computeFlag(col("message"))).withColumn("flag3",computeFlag(col("message")))

方法 2

 val tempDF = df.withColumn("flagValues",computeMultipleFlags(col("message"))).withColumn("_tmp", split($"flagValues","#")).select($"message",$"_tmp".getItem(0).as("flag1"),$"_tmp".getItem(1).as("commercial"),$"_tmp".getItem(2).as("flag2"),$"_tmp".getItem(3).as("flag3")).drop("_tmp")

UDF : computeFlag 根据各个关键字列表的精确匹配返回 1 或 0

UDF : computeMultipleFlags 根据 flag1、flag 2 和 flag 3 各自关键字的精确匹配返回 # 分隔结果 1 或 0：示例 1#0#0

我已经使用这两种方法解决了，但是看到/感觉方法 2 表现更好。请指教。

• Spark 数据帧默认是并行化的，但是这种情况如何 采用方法 1。flag1、flag2、flag3 列将在 并行还是顺序？

• Spark 数据框会自动并行处理我的输入列吗 “消息”：针对列的多行多线程
  计算？

Answer 1

在这两种情况下，您都使用了udf 函数，这需要column serialization 和deserialization，在处理过程中花费了宝贵的时间。

在您的第一种情况下，您使用 withColumn 调用了相同的 udf 三次，这意味着每次生成标志都会发生三次序列化和反序列化

在第二种情况下，您只定义了一个 udf 函数。所以很明显比第一个跑得更快更高效。而且你已经使用了split 函数，这是一个好兆头。

Dataframes 在本质上是分布式的，因此每个函数调用在每个executors 中并行执行，但每个函数将按顺序执行，即数据并行化，但函数/任务仍然是连续的。

希望解释清楚

使用Spark functions 可以为您的情况提供更好的解决方案，它不需要像udf 情况那样进行额外的序列化和反序列化。您可以使用以下解决方案。

df.withColumn("_tmp", split($"message", " "))
  .select($"message",
    when(array_contains($"_tmp", "stack") || array_contains($"_tmp", "slack"), "1").otherwise("0") as "flag1",
    when(array_contains($"_tmp", "twitter") || array_contains($"_tmp", "facebook") || array_contains($"_tmp", "whatsapp"), "1").otherwise("0") as "flag2",
    when(array_contains($"_tmp", "flow") || array_contains($"_tmp", "run") || array_contains($"_tmp", "increase"), "1").otherwise("0") as "flag3")
  .show(false)