Walt You - 行是知之始

Learning Spark (0) - 总览及第三章结构化API

2020-09-08
Spark
   

Databricks 出了本《Learning Spark》的电子书，来学习一下。

总览

可以看出这本书总共分为五大部分，分别为：

  1. 基本介绍
  2. Spark SQL 使用与调优
  3. Streaming
  4. MLlib
  5. Spark 3.0 的新特性

电子书下载地址： https://databricks.com/p/ebook/learning-spark-from-oreilly

前两章非常基础，直接从第三章《Structured APIs》开始。

RDD 到底是什么？

RDD 是 Spark 最基础的抽象。它分为三部分：

  1. Dependencies 依赖
  2. Partitions 分区
  3. Compute funtion: Partitions => Iterator[T] 计算函数

第一部分：依赖。定义了这个RDD是从何处来的，也保证了必要时重新计算这个RDD。第二部分：partitions 分区。将数据分区存储，天然支持并行计算。第三部分计算函数可以将消费分区数据，然后得到我们想要的 Iterator[T].

整体架构简洁明了。但是也存在几个问题。其中一个就是compute function对Spark是不透明的。对于Spark来讲，compute function 只是一个 lambda expression，不管它实际上是join、filter、select。 另外一个问题是 Iterator[T] 的数据类型，对Python RDD来讲，也是透明的，Spark只知道它是一个python的 generic object。

而且，因为无法检查计算函数，那就无从谈起优化这个函数。另外因为不知道数据类型，也无法直接获取某种类型的某列，Spark 只能将它序列化为字节，而且不能使用任何压缩技术。

毫无疑问，这些不透明性，影响了Spark优化我们的计算，那么解决方案是什么呢？

结构化Spark

Spark 2.x 为结构化Spark提供了几个关键方案。

一个是在数据分析中找到的一些通用模式来表达计算。这些模式表示为high-level 处理，比如有filtering、selecting、counting、aggregating、averaging 和 grouping。这样子更清晰和简单。

通过在DSL中使用一组通用运算符，可以进一步缩小这种特异性。 通过DSL中的一组操作（可以在Spark受支持的语言（Java，Python，Spark，R和SQL）中以API的形式使用），这些运算符可以让您告诉Spark您希望对数据进行什么计算，结果， 它可以构建一个有效的执行查询计划。

最终的结构方案是允许我们以表格格式（如SQL表格或电子表格）排列数据，并使用受支持的结构化数据类型。 但是，这种结构有什么用呢？

主要优点和好处

  • expressivity 表现力
  • simplicity 简单性
  • composability 可组合性
  • uniformity 统一性

使用low-level RDD API：

# In Python
# Create an RDD of tuples (name, age)
dataRDD = sc.parallelize([("Brooke", 20), ("Denny", 31), ("Jules", 30), ("TD", 35)， ("Brooke", 25)])
# Use map and reduceByKey transformations with their lambda # expressions to aggregate and then compute average
agesRDD = (dataRDD
           .map(lambda x: (x[0], (x[1], 1)))
           .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))
           .map(lambda x: (x[0], x[1][0]/x[1][1])))

这堆lambda 函数，神秘且难读懂。同时Spark也不了解这些代码的意图。而且相同意图下，用Scala写的代码看起来也和用Python写的有很大不同。

high-level API：

# In Python
from pyspark.sql import SparkSession 
from pyspark.sql.functions import avg 
# Create a DataFrame using SparkSession 
spark = (SparkSession
         .builder
	       .appName("AuthorsAges")
         .getOrCreate())
# Create a DataFrame
data_df = spark.createDataFrame([("Brooke", 20), ("Denny", 31), ("Jules", 30), ("TD", 35), ("Brooke", 25)], ["name", "age"])
# Group the same names together, aggregate their ages, and compute an average 
avg_df = data_df.groupBy("name").agg(avg("age"))
# Show the results of the final execution
avg_df.show()

+------+--------+
|  name|avg(age)|
+------+--------+
|Brooke|    22.5|
| Jules|    30.0|
|    TD|    35.0|
| Denny|    31.0|
+------+--------+

不过使用high-level、表达力强的DSL 操作符，还是会有一些争议（虽然这些操作符在数据分析中很常见或是重复率很高），那就是我们限制了开发人员的控制他们请求的能力。但是请放心，Spark支持你随时切换到low-level的API中，虽然几乎没有必要。

除了易于阅读之外，Spark的高级API的结构还引入了其组件和语言之间的统一性。 例如，此处显示的Scala代码与以前的Python代码具有相同的作用-并且API看起来几乎相同：

// In Scala
import org.apache.spark.sql.functions.avg 
import org.apache.spark.sql.SparkSession 
// Create a DataFrame using SparkSession 
val spark = SparkSession
      .builder
      .appName("AuthorsAges")
      .getOrCreate()
// Create a DataFrame of names and ages
val dataDF = spark.createDataFrame(Seq(("Brooke", 20), ("Brooke", 25), ("Denny", 31), ("Jules", 30), ("TD", 35))).toDF("name", "age")
// Group the same names together, aggregate their ages, and compute an average
val avgDF = dataDF.groupBy("name").agg(avg("age")) 
// Show the results of the final execution
avgDF.show()

正是因为Spark 构建了high-level结构化API的Spark SQL引擎，这种简单性和表达性才成为可能。由于有了此引擎，它是所有Spark组件的基础，所以我们获得了统一的API。 无论您是在结构化流式处理还是MLlib中对DataFrame表达查询，您始终都在将DataFrame转换为结构化数据并对其进行操作。

DataFrame API

受 pandas dataframe 的启发，Spark Dataframe 就像是一张分布在内存中的表格，它带有column name 和type。

基本数据类型

结构化的和复杂的数据类型

Schemas 和create DataFrame

Spark中的schema定义了DataFrame的列名和关联的数据类型。

相对于采用“读取schema”方法，预先定义schema具有三个好处：

  • 您可以把 Spark 从推断数据类型的负担中释放出来。
  • 您可以阻止Spark创建单独的作业，而只是为了读取文件的大部分内容以确定schema，这对于大型数据文件而言可能既昂贵又费时。
  • 您可以及早发现错误，如果数据与架构不匹配。

定义 schema

Using Spark DataFrame API:

// In Scala
import org.apache.spark.sql.types._
val schema = StructType(Array(StructField("author", StringType, false),StructField("title", StringType, false), StructField("pages", IntegerType, false)))

# In Python
from pyspark.sql.types import *
schema = StructType([StructField("author", StringType(), False),
      StructField("title", StringType(), False),
      StructField("pages", IntegerType(), False)])

Using DDL:

// In Scala
val schema = "author STRING, title STRING, pages INT" 

# In Python
schema = "author STRING, title STRING, pages INT"

df = spark.createDataFrame(data, schema)

列和表达式

Spark 也支持在column上进行逻辑转换：expr(“columnName * 5”) or (expr(“colum nName - 5”) > col(anothercolumnName)) 。

scala> import org.apache.spark.sql.functions._
scala> blogsDF.columns
res2: Array[String] = Array(Campaigns, First, Hits, Id, Last, Published, Url)
// Access a particular column with col and it returns a Column type
scala> blogsDF.col("Id")
res3: org.apache.spark.sql.Column = id
// Use an expression to compute a value
scala> blogsDF.select(expr("Hits * 2")).show(2) 
// or use col to compute value
scala> blogsDF.select(col("Hits") * 2).show(2)

// Use an expression to compute big hitters for blogs
// This adds a new column, Big Hitters, based on the conditional expression 
blogsDF.withColumn("Big Hitters", (expr("Hits > 10000"))).show()

// Concatenate three columns, create a new column, and show the 
// newly created concatenated column
blogsDF
.withColumn("AuthorsId", (concat(expr("First"), expr("Last"), expr("Id")))) .select(col("AuthorsId"))
.show(4)

// These statements return the same value, showing that 
// expr is the same as a col method call 
blogsDF.select(expr("Hits")).show(2)
blogsDF.select(col("Hits")).show(2) 
blogsDF.select("Hits").show(2)

// Sort by column "Id" in descending order 
blogsDF.sort(col("Id").desc).show() 
blogsDF.sort($"Id".desc).show()

Rows 行

创建行

// In Scala
import org.apache.spark.sql.Row
// Create a Row
val blogRow = Row(6, "Reynold", "Xin", "https://tinyurl.6", 255568, "3/2/2015",
Array("twitter", "LinkedIn"))
// Access using index for individual items blogRow(1)
res62: Any = Reynold

# In Python
from pyspark.sql import Row
blog_row = Row(6, "Reynold", "Xin", "https://tinyurl.6", 255568, "3/2/2015",
["twitter", "LinkedIn"])
# access using index for individual items 
blog_row[1]
'Reynold'

Create DataFrame by rows：

// In Scala
val rows = Seq(("Matei Zaharia", "CA"), ("Reynold Xin", "CA")) 
val authorsDF = rows.toDF("Author", "State")
authorsDF.show()

# In Python
rows = [Row("Matei Zaharia", "CA"), Row("Reynold Xin", "CA")]
authors_df = spark.createDataFrame(rows, ["Authors", "State"])
authors_df.show()

通用的DataFrame操作

Spark 提供 DataFrameReader 来读取各式各样的数据源来生成DataFrame，比如 JSON, CSV, Parquet, Text, Avro, ORC, etc。 同时，也提供 DataFrameWriter 来把 DataFame 写回各类数据源。

使用 DataFrameReader & DataFrameWriter

在Spark中读取和写入非常简单，因为社区提供了这些高级抽象和贡献，可以连接到各种数据源，包括常见的NoSQL存储，RDBMS，Apache Kafka和Kinesis等流引擎。

Without schema:

// In Scala
val sampleDF = spark 
	.read
	.option("samplingRatio", 0.001)
	.option("header", true) 
	.csv("""/databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv""")

Define schema firstly:

# In Python, define a schema
from pyspark.sql.types import *
# Programmatic way to define a schema
fire_schema = StructType([StructField('CallNumber', IntegerType(), True),
                          StructField('UnitID', StringType(), True),
                          StructField('IncidentNumber', IntegerType(), True),
                          StructField('CallType', StringType(), True),
                          StructField('CallDate', StringType(), True),
                          StructField('WatchDate', StringType(), True),
                          ...
                          ...
                          StructField('Delay', FloatType(), True)])
# Use the DataFrameReader interface to read a CSV file
sf_fire_file = "/databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv"
fire_df = spark.read.csv(sf_fire_file, header=True, schema=fire_schema)

// In Scala it would be similar
val fireSchema = StructType(Array(StructField("CallNumber", IntegerType, true),
                                  StructField("Location", StringType, true),
                                  ...
                                  ...
                                  StructField("Delay", FloatType, true)))
// Read the file using the CSV DataFrameReader
val sfFireFile="/databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv" 
val fireDF = spark.read.schema(fireSchema)
      .option("header", "true")
      .csv(sfFireFile)

Saving a DataFrame as a Parquet file or SQL table

你可以使用 DataFrameWriter 保持 DataFrame 到各类数据源中。 Parquet，是默认的格式。这种格式使用snappy来压缩数据，并且会将schema信息保存到metadata中，这样子下次读取这个文件的时候，就不需要手动提供schema。

保存为Parquet 文件：

// In Scala to save as a Parquet file
val parquetPath = ... 
fireDF.write.format("parquet").save(parquetPath)

# In Python to save as a Parquet file
parquet_path = ...
fire_df.write.format("parquet").save(parquet_path)

保存为sql table，这个表会注册到Hive的metastore 中：

// In Scala to save as a table
val parquetTable = ... // name of the table fireDF.write.format("parquet").saveAsTable(parquetTable)

# In Python
parquet_table = ... # name of the table fire_df.write.format("parquet").saveAsTable(parquet_table)

Projections and filters

关系表达式中的projection是一种通过使用过滤器 filters 仅返回与特定关系条件匹配的行的方法。 在Spark中，投影是使用 select() 方法完成的，而过滤器可以使用 filter() 或 where() 方法来表示。

# In Python
few_fire_df = (fire_df
               .select("IncidentNumber", "AvailableDtTm", "CallType")
               .where(col("CallType") != "Medical Incident"))
few_fire_df.show(5, truncate=False)

// In Scala
val fewFireDF = fireDF
	.select("IncidentNumber", "AvailableDtTm", "CallType") 
	.where($"CallType" =!= 	"Medical Incident")
fewFireDF.show(5, false)

重命名、添加、删除列

# In Python
fire_ts_df = (new_fire_df
              .withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy"))
              .drop("CallDate")
              .withColumn("OnWatchDate", to_timestamp(col("WatchDate"), "MM/dd/yyyy"))
              .drop("WatchDate")
              .withColumn("AvailableDtTS", to_timestamp(col("AvailableDtTm"),
                                                        "MM/dd/yyyy hh:mm:ss a"))
              .drop("AvailableDtTm"))
# Select the converted columns
(fire_ts_df
 .select("IncidentDate", "OnWatchDate", "AvailableDtTS")
 .show(5, False))


// In Scala
val fireTsDF = newFireDF
  .withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy"))
  .drop("CallDate")
  .withColumn("OnWatchDate", to_timestamp(col("WatchDate"), "MM/dd/yyyy"))
  .drop("WatchDate")
  .withColumn("AvailableDtTS", to_timestamp(col("AvailableDtTm"), "MM/dd/yyyy hh:mm:ss a"))
  .drop("AvailableDtTm")
// Select the converted columns
fireTsDF
	.select("IncidentDate", "OnWatchDate", "AvailableDtTS") .show(5, false)

聚合 Aggregations

DataFrame 有很多有用的 transforma 和 action，比如 groupBy(), orderBy(), and count()，他们提供了通过列名进行聚合然后得出数量的能力。

# In Python
(fire_ts_df
 .select("CallType")
 .where(col("CallType").isNotNull())
 .groupBy("CallType")
 .count()
 .orderBy("count", ascending=False)
 .show(n=10, truncate=False))

// In Scala
fireTsDF
  .select("CallType") 
  .where(col("CallType").isNotNull) 
  .groupBy("CallType")
  .count()
  .orderBy(desc("count")) 
  .show(10, false)

其他操作

除了上面看到的，DataFrame API 还提供了状态统计函数：min(), max(), sum(), and avg()。

# In Python
import pyspark.sql.functions as F 
(fire_ts_df
      .select(F.sum("NumAlarms"), F.avg("ResponseDelayedinMins"),
              F.min("ResponseDelayedinMins"), F.max("ResponseDelayedinMins"))
      .show())

// In Scala
import org.apache.spark.sql.{functions => F} 
fireTsDF
      .select(F.sum("NumAlarms"), F.avg("ResponseDelayedinMins"),
      	F.min("ResponseDelayedinMins"), F.max("ResponseDelayedinMins"))
      .show()

Datasets API

Spark 2.0 统一了 DataFrame 和 Dataset 的API，称为 Structured API，以便开发者只需要学习一套API。Dataset 有两个部分：Untyped API和Typed API：

从概念上来讲，Scala 中的 DataFrame 可以被当作是一个普通对象的合集：Dataset[Row]。Row 是一个通用的、没有类型的 JVM 对象，它可以包含多个不同类型的字段。

不同的是，一个 Dataset 是一个包含强类型对象的合集，这些对象可以是 Java 中的类或者 Scala 中的JVM object。

类型对象、无类型对象和通用的Row

在 Spark 支持的语言当中， Dataset 只在 Java 、Scala 中有意义，而在 Python、R 中只有 DataFrame 有意义。这是因为 Python、R 不是编译时间类型安全的（compile-time type-safe），它的类型是在执行期间而不是在编译期间，动态推断或分配的。

在Scala和Java中则相反：在编译时将类型绑定到变量和对象。 但是，在Scala中，DataFrame只是无类型Dataset [Row]的别名。

创建 Dataset

与从数据源创建 DataFrame 一样，在创建数据集时，您必须了解shema。换句话说，您需要了解数据类型。尽管使用 JSON 和 CSV 数据可以推断架构，但对于大型数据集，这需要大量资源（非常昂贵）。在 Scala 中创建数据集时，为生成的数据集指定架构的最简单方法是使用case class。在 Java 中，使用 JavaBean 类。

Scala： Case classes

JSON file 中每一行大概是这样子：

{"device_id": 198164, "device_name": "sensor-pad-198164owomcJZ", "ip":
    "80.55.20.25", "cca2": "PL", "cca3": "POL", "cn": "Poland", "latitude":
    53.080000, "longitude": 18.620000, "scale": "Celsius", "temp": 21,
    "humidity": 65, "battery_level": 8, "c02_level": 1408,"lcd": "red",
    "timestamp" :1458081226051}

怎么把它变成 typed object：DeviceIoTData 呢？我们可以定义一个case class：

case class DeviceIoTData (battery_level: Long, c02_level: Long, cca2: String, cca3: String, cn: String, device_id: Long, device_name: String, humidity: Long, ip: String, latitude: Double, lcd: String, longitude: Double, scale:String, temp: Long, timestamp: Long)

定义之后，我们就可以直接读取文件并将返回类型 Dataset[Row] 转换为 Dataset[DeviceIoTData]：

// In Scala
val ds = spark.read 
	.json("/databricks-datasets/learning-spark-v2/iot-devices/iot_devices.json")
  .as[DeviceIoTData]
ds: org.apache.spark.sql.Dataset[DeviceIoTData] = [battery_level...]
ds.show(5, false)

Dataset 的操作

和 DataFrame 类似，dataset也可以进行 transformations and actions。

// In Scala
val filterTempDS = ds.filter({d => {d.temp > 30 && d.humidity > 70}) filterTempDS: org.apache.spark.sql.Dataset[DeviceIoTData] = [battery_level...] filterTempDS.show(5, false)

需要注意的另一件事是，使用 DataFrames 时，fitler 像是 SQL-like 的 DSL 语言一样，但是在 Dataset 中，我们只能使用scala 或 java的原生语言表达。

// In Scala
case class DeviceTempByCountry(temp: Long, device_name: String, device_id: Long, cca3: String)
val dsTemp = ds
	.filter(d => {d.temp > 25})
	.map(d => (d.temp, d.device_name, d.device_id, d.cca3)) 
	.toDF("temp", "device_name", "device_id", "cca3")
	.as[DeviceTempByCountry] 

dsTemp.show(5, false)


val dsTemp2 = ds
	.select($"temp", $"device_name", $"device_id", $"device_id", $"cca3") 
	.where("temp > 25")
	.as[DeviceTempByCountry]

总结一下，我们在 Dataset上进行的操作，比如 filter(), map(), groupBy(), select(), take(), etc， 和DataFrame上的很类似。 某种程度上来讲，Dataset 和 RDD 很像，它们都提供相似的接口给上述的方法以及编译时期的安全，但是Dataset的接口更安全和面向对象。

当我们使用 dataset时， Spark SQL 引擎处理 JVM object 的创建、转换、序列化和反序列化。它会在 Dataset encoder的帮助下，处理 Java 堆内存管理。

DataFrame VS Dataset

现在，您可能想知道为什么以及何时应该使用DataFrames或Dataset。 在许多情况下，哪种方法都可以，取决于您使用的语言，但是在某些情况下，一种语言比另一种语言更可取。 这里有一些例子：

  • 如果您想告诉Spark该做什么而不是该怎么做，请使用DataFrames或Datasets。
  • 如果您想要丰富的语义，高级抽象和DSL运算符，请使用DataFrames 或 Datasets。
  • 如果您希望严格的编译时类型安全性，并且不介意为特定的数据集[T]创建多个case class，请使用dataset。
  • 如果您的处理需要高级表达式，过滤器，映射，聚合，计算平均值或总和，SQL查询，列访问或对半结构化数据使用关系运算符，请使用DataFrames或Datasets。
  • 如果您的处理要求与类似SQL的查询类似的关系转换，请使用DataFrames。
  • 如果您想利用Tungsten的高效编码器序列化并从中受益，请使用Dataset。
  • 如果要跨Spark组件进行统一，代码优化和API简化，请使用DataFrames。
  • 如果您是R用户，请使用DataFrames。
  • 如果您是Python用户，则在需要更多控制权时，请使用DataFrames并下拉至RDD。
  • 如果需要空间和速度效率，请使用DataFrames。
  • 如果要在编译期间而不是在运行时捕获错误，请选择适当的API，如下图所示：

什么时候使用RDD

您可能会问：RDD是否被降级为二等公民？ 他们被弃用了吗？ 答案是否定的！ 尽管将来在Spark 2.x和Spark 3.0中进行的所有开发工作都将继续具有DataFrame接口和语义，而不是使用RDD，但仍将继续支持RDD API。

在某些情况下，您可能需要考虑使用RDD，例如：

  • 使用RDD编写的第三方程序包
  • 可以放弃代码优化、高效的空间利用以及DataFrame和Dataset所提供的性能优势
  • 想精确地指导Spark如何进行查询

此外，您可以使用简单的API方法调用df.rdd随意在DataFrame或Dataset和RDD之间无缝切换。 （但是，请注意，这个操作是有代价的，除非有必要，否则应避免。）毕竟，DataFrame和Dataset是建立在RDD之上的，它们在整个代码生成过程中都分解为紧凑的RDD代码。

构建高效查询并生成紧凑代码的过程是Spark SQL引擎的工作。 这是我们一直在研究的结构化API的基础。 现在让我们来窥探一下该引擎。

有个视频也讲述了 RDD/dataFrame/dataset之间的区别，可以去看一看：A Tale of Three Apache Spark APIs: RDDs, DataFrames, and Datasets - Jules Damji 。

Spark SQL 和底层引擎

Spark 1.3中引入以来，Spark SQL已经发展成为一个强大的引擎，并在其上构建了许多高级结构化功能。 除了允许您对数据发出类似SQL的查询之外，Spark SQL引擎还支持：

  • 统一Spark组件，并允许抽象为Java，Scala，Python和R中的DataFrame /dataset，从而简化了结构化数据集的工作。
  • 连接到Apache Hive metastore和table。
  • 从结构化文件格式（JSON，CSV，文本，Avro，Parquet，ORC等）中以特定模式读取和写入结构化数据，并将数据转换为临时表。
  • 提供交互式Spark SQL Shell，以快速浏览数据。
  • 通过标准数据库JDBC / ODBC连接器提供与外部工具之间的桥梁。
  • 生成用于JVM的优化查询计划和紧凑代码，以最终执行。

Spark SQL引擎的核心是Catalyst优化器和Project Tungsten。 它们一起支持高级DataFrame和Dataset API和SQL查询。 现在，让我们仔细看看优化器。

The Catalyst Optimizer

Catalyst Optimizer 将可执行的查询转换为一个可执行计划。它通过以下四个阶段：

  1. 分析 Analysis
  2. 逻辑优化 Logical optimization
  3. 物理计划 Physical planning
  4. 产生代码 Code generation

举个例子，不管你是用什么语言，都是类似的处理过程：从query plan 到 execute bytecode。

# In Python
count_mnm_df = (mnm_df
                .select("State", "Color", "Count")
                .groupBy("State", "Color")
                .agg(count("Count").alias("Total"))
                .orderBy("Total", ascending=False))

-- In SQL
SELECT State, Color, Count, sum(Count) AS Total FROM MNM_TABLE_NAME
GROUP BY State, Color, Count
ORDER BY Total DESC

在Python中，执行 count_mnm_df.explain(True) 可以查看各个阶段的计划。或者想要看逻辑计划和物理计划之间的不同，可以在scala中使用df.queryExecution.logical or df.queryExecution.optimizedPlan 。

count_mnm_df.explain(True)

== Parsed Logical Plan ==
'Sort ['Total DESC NULLS LAST], true
+- Aggregate [State#10, Color#11], [State#10, Color#11, count(Count#12) AS...]
+- Project [State#10, Color#11, Count#12]
+- Relation[State#10,Color#11,Count#12] csv

== Analyzed Logical Plan ==
State: string, Color: string, Total: bigint
Sort [Total#24L DESC NULLS LAST], true
+- Aggregate [State#10, Color#11], [State#10, Color#11, count(Count#12) AS...]
+- Project [State#10, Color#11, Count#12]
+- Relation[State#10,Color#11,Count#12] csv

== Optimized Logical Plan ==
Sort [Total#24L DESC NULLS LAST], true
+- Aggregate [State#10, Color#11], [State#10, Color#11, count(Count#12) AS...]
+- Relation[State#10,Color#11,Count#12] csv

== Physical Plan ==
*(3) Sort [Total#24L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(Total#24L DESC NULLS LAST, 200)
+- *(2) HashAggregate(keys=[State#10, Color#11], functions=[count(Count#12)], output=[State#10, Color#11, Total#24L])
+- Exchange hashpartitioning(State#10, Color#11, 200) +- *(1) HashAggregate(keys=[State#10, Color#11],
functions=[partial_count(Count#12)], output=[State#10, Color#11, count#29L]) +- *(1) FileScan csv [State#10,Color#11,Count#12] Batched: false,
Format: CSV, Location: InMemoryFileIndex[file:/Users/jules/gits/LearningSpark2.0/chapter2/py/src/... dataset.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<State:string,Color:string,Count:int>

让我们考虑另一个DataFrame计算示例。 随着底层引擎优化其逻辑和物理计划，以下Scala代码也经历了类似的旅程：

// In Scala
// Users DataFrame read from a Parquet table val usersDF = ...
// Events DataFrame read from a Parquet table val eventsDF = ...
// Join two DataFrames
val joinedDF = users
      .join(events, users("id") === events("uid"))
      .filter(events("date") > "2015-01-01")

阶段1: 分析

Spark SQL引擎首先为SQL或DataFrame查询生成抽象语法树（AST）。 在此初始阶段，将通过查询内部catalog（Spark SQL的编程接口，其中包含列，数据类型，函数，表，数据库等的名称的列表）来解析任何列或表的名称。 成功解决后，查询将进入下一个阶段。

阶段2: 逻辑优化

这个阶段包括两个内部阶段。 应用基于标准规则（RBO）的优化方法，Catalyst优化器将首先构建一组多个计划，然后使用其基于成本的优化器（CBO）将成本分配给每个计划。 这些计划以operator trees的形式布置； 例如，它们可能包括恒定折叠，谓词下推，投影修剪，布尔表达式简化等过程。此逻辑计划是物理计划的输入。

阶段3: 物理计划

在此阶段，Spark SQL使用与Spark执行引擎中可用的物理运算符匹配的物理运算符，为选定的逻辑计划生成最佳的物理规划。

阶段4: 生成代码

查询优化的最后阶段涉及生成可在每台计算机上运行的有效Java字节码。 因为Spark SQL可以对内存中加载的数据集进行操作，所以Spark可以使用最新的编译器技术来生成代码以加快执行速度。 换句话说，它充当编译器。 Tungsten项目在此起到了重要作用，该项目可促进整个阶段的代码生成。 整个阶段的代码生成是什么？ 这是一个物理查询优化阶段，它将整个查询分解为一个函数，摆脱虚拟函数调用，并使用CPU寄存器存储中间数据。 Spark 2.0中引入的第二代Tungsten引擎使用此方法生成紧凑的RDD代码以最终执行。 这种简化的策略大大提高了CPU效率和性能。

Similar Posts

上一篇 VSCode LSP Extension 指南

下一篇 Learning Spark (1) - Spark SQL 和 DataFrame：内置数据源

Content