ColumnarToRow 如何在 Spark 中进行高效操作答案

【问题标题】：How is ColumnarToRow an efficient operation in SparkColumnarToRow 如何在 Spark 中进行高效操作
【发布时间】：2021-02-23 18:19:43
【问题描述】：

据我了解，列格式更适合 MapReduce 任务。即使对于某些列的选择，列式也能很好地工作，因为我们不必将其他列加载到内存中。

但在 Spark 3.0 中，我看到在查询计划中应用了这个 ColumnarToRow 操作，据我所知，docs 将数据转换为行格式。

它比柱状表示更有效，有哪些见解可以控制此规则的应用？

对于以下代码，我附上了查询计划。

import pandas as pd

df = pd.DataFrame({
    'a': [i for i in range(2000)],
    'b': [i for i in reversed(range(2000))],
})

df = spark.createDataFrame(df)

df.cache()
df.select('a').filter('a > 500').show()

【问题讨论】：

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

【解决方案1】：

我只是简要地阅读了这一点，但似乎这个一般逻辑成立：

列格式可帮助您最有效地选择某些列。行格式可帮助您最有效地选择某些行。

因此，当您想大量选择某些行时，例如每个国家/地区，每天，...' 使用基于行的格式并在过滤器列上有一个索引通常是有意义的。

这里还有一个参考，它们表明要定义一个索引：https://spark.apache.org/docs/3.0.0/sql-ref-syntax-qry-select-hints.html

【讨论】：

【解决方案2】：

我的猜测是，这个 WSCG 中的 ColumnarToRow 部分实际上是 pandas 数据帧到 Spark 数据帧的转换，而不是任何关于 Spark 如何处理自己的数据帧的指示。

如果我们从“原生”Spark df 开始，计划看起来会大不相同：

>>> a = range(2000)                                                                                                                                                    
>>> b = [ i for i in reversed(range(2000))]                                                                                                                            
>>> df = spark.createDataFrame(zip(a,b),["a","b"])                                                                                                 
>>> df.select('a').filter('a > 500').show()

此外，您所指的链接说：

case classColumnarToRowExec(child: SparkPlan) extends SparkPlan with UnaryExecNode with CodegenSupport with Product with Serializable

提供一个通用的执行器来将 ColumnarBatch 的 RDD 翻译成 InternalRow 的 RDD。每当这样的转换是插入确定需要。

...这基本上意味着将外部 RDD（您的情况下的熊猫）转换为内部 Spark 表示（InternalRows 的 RDD）。

【讨论】：