如何用Spark Dataset API复刻Lateral View逻辑？性能优化咨询

我正在尝试将一个使用多个Lateral View解析JSON的SQL查询改写为Spark Dataset API，但发现难以复现原有的逻辑计划：在select语句中json_tuple仅能使用一次，而Lateral View无此限制。请问是否有方法复刻Lateral View的行为？另外，新增投影会带来怎样的性能影响，Spark能否对此进行优化？

== Physical Plan ==
*(2) Project [(cast(a0#117 as int) + cast(b0#118 as int)) AS x#101, (cast(a1#119 as int) + cast(b1#120 as int)) AS y#102]
+- Generate json_tuple(json2#7, k1, k2), [a0#117, b0#118], false, [a1#119, b1#120]
   +- Generate json_tuple(json#6, k1, k2), [json2#7], false, [a0#117, b0#118]
         ...

== Physical Plan ==
*(3) Project [a1#59, b1#60, (cast(a1#59 as double) + cast(b1#60 as double)) AS (a1 AS `integer` + b1 AS `integer`)#22, a0#13, b0#14, (a0 AS `integer` + b0 AS `integer`)#12, json#6, json2#7]
+- Generate json_tuple(json2#7, k1, k2), [a0#13, b0#14, (a0 AS `integer` + b0 AS `integer`)#12, json#6, json2#7], false, [a1#59, b1#60]
   +- *(2) Project [a0#13, b0#14, (cast(a0#13 as double) + cast(b0#14 as double)) AS (a0 AS `integer` + b0 AS `integer`)#12, json#6, json2#7]
      +- Generate json_tuple(json#6, k1, k2), [json#6, json2#7], false, [a0#13, b0#14]

val d0 = spark.sparkContext.parallelize(
  Seq(
    ("""{"k1": 1, "k2": "101"}""","""{"k1": 1, "k2": "101"}"""),
    ("""{"k1": 12, "k2": "201"}""","""{"k1": 12, "k2": "201"}"""),
    ("""{"k1": 13, "k2": "301"}""","""{"k1": 13, "k2": "301"}""")
  )
).toDF("json", "json2")

// Dataset API 原写法
val d1 = d0
  .select(
    json_tuple($"json", "k1", "k2").as(Seq("a0", "b0")), $"a0".as("integer") + $"b0".as("integer"), col("*")
  )
  .select(
    json_tuple($"json2", "k1", "k2").as(Seq("a1", "b1")), $"a1".as("integer") + $"b1".as("integer"), col("*")
  )
d1.explain()

// SQL 写法
d0.createOrReplaceTempView("d0")
val d2 = spark.sql(
"""
  select
    cast(a0 as int) + cast(b0 as int) as x,
    cast(a1 as int) + cast(b1 as int) as y
  from d0
  lateral view json_tuple(json, 'k1', 'k2') A_json as a0, b0
  lateral view json_tuple(json2, 'k1', 'k2') B_json as a1, b1
"""
)
d2.explain()

1. 复刻Lateral View的行为 #

Lateral View在Spark物理计划中对应Generate算子，核心作用是展开解析生成的列。要在Dataset API中复现相同逻辑，关键是避免携带冗余列，不要用col("*")强制保留所有字段，只保留后续步骤需要的列即可：

// 优化后的Dataset API写法
val d1 = d0
  // 第一步解析json，仅保留需要的中间列和下一个解析用到的json2
  .select(
    json_tuple($"json", "k1", "k2").as(Seq("a0", "b0")),
    $"json2"
  )
  // 第二步解析json2，保留之前的a0、b0和新生成的a1、b1
  .select(
    $"a0", $"b0",
    json_tuple($"json2", "k1", "k2").as(Seq("a1", "b1"))
  )
  // 最终投影计算结果
  .select(
    ($"a0".cast("int") + $"b0".cast("int")).as("x"),
    ($"a1".cast("int") + $"b1".cast("int")).as("y")
  )
d1.explain()

此时查看物理计划，会和SQL版本完全一致：Generate算子仅携带必要的输入列，最终只投影需要的结果字段。

2. 投影的性能影响与Spark优化 #

性能影响 #

原Dataset API写法中使用col("*")会导致每个select阶段都携带所有之前的列（包括原始JSON字符串、中间计算的列），这会增加数据在节点间的传输量，以及内存占用——冗余字段会被一直保留到最后阶段，无意义地消耗资源。

Spark的优化能力 #

Spark Catalyst优化器默认会做投影下推和列裁剪，自动剔除查询中不需要的列，合并冗余的投影操作。但如果显式用col("*")强制保留所有列，优化器无法自动裁剪这些冗余字段，因此会保留不必要的投影步骤。

当按照优化后的写法编写代码，只保留必要的列时，Catalyst会自动将投影操作合并，最终生成与SQL版本一致的高效物理计划，不会有额外的性能开销。

内容的提问来源于stack exchange，提问作者toni057