结构化流DataFrame操作

前面的例子表明，一旦配置和定义了数据源，DataStreamReader将返回一个DataFrame的实例。这意味着我们可以使用大多数熟悉的操作和PySpark SQL函数来表达应用程序流计算逻辑。但是要注意，并不是所有的DataFrame操作都受流式DataFrame支持的，比如limit、distinct和sort就不能在流DataFrame上使用，这是因为它们在流数据处理的上下文中不适用。

选择、投影和聚合操作

结构化流的一个优点是具有一组用于PySpark的批处理和流处理的统一API。使用流数据格式的DataFrame，可以应用任何select和filter转换，以及任何作用在个别列上的PySpark SQL函数。此外，基本聚合和高级分析函数也可用于流DataFrame。

【示例】移动电话事件数据流分析。

移动电话的开关机等事件会保存在json格式的文件中。现在编写PySpark结构化流处理程序来读取这些事件并处理。请按以下步骤操作。

1）准备数据

在本示例中，我们使用文件数据源，该数据源以json文件的格式记录了一小组移动电话动作事件。每个事件由三个字段组成：

id：表示手机的唯一ID。在样例数据集中，电话ID将类似于phone1、phone2、phone3等。
action：表示用户所采取的操作。该操作的可能值是"open"或"close"。
ts：表示用户action发生时的时间戳。这是事件时间(event time)。

我们准备了三个存储移动电话事件数据的JSON文件：file1.json, file2.json, file3.json。这三个文件位于PBLP平台的~/data/spark/mobile目录下。

为了模拟数据流的行为，我们将把这三个JSON文件复制到项目的“src/main/data/mobile”目录下。

2）先导入相关的依赖包，并构造一个SparkSession实例，代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *

# 创建SparkSession实例
spark = SparkSession \
        .builder \
        .appName("streaming demo") \
        .getOrCreate()

# 设置shuffle后的分区数为10（测试环境下）
spark.conf.set("spark.sql.shuffle.partitions",10)

3）为手机事件数据创建模式（schema）

默认情况下，结构化流在从基于文件的数据源读取数据时需要一个模式（因为最初目录可能是空的，因此结构化的流无法推断模式）。但是，可以设置配置参数spark.sql.streaming.schemaInference的值为true来启用模式推断。在这个例子中，我们将显式地创建一个模式，代码如下所示：

# 为手机事件数据创建一个schema
fields = [
      StructField("id", StringType(), nullable = False),
      StructField("action", StringType(), nullable = False),
      StructField("ts", TimestampType(), nullable = False)
]
mobileDataSchema = StructType(fields)

3）读取流文件数据源，创建DataFrame，并将action列值转换为大写，代码如下：

# 指定监听的文件目录
dataPath = "/data/spark/stream/mobile"

# 读取指定目录下的源数据文件，一次一个
mobileDF = spark.readStream \
      .option("maxFilesPerTrigger", 1) \
      .option("mode","failFast") \
      .schema(mobileDataSchema) \
      .json(dataPath)

# mobileSSDF.printSchema()

4）将action列值转换为大写，执行过滤、投影、聚合等转换操作，代码如下：

mobileDF2 = mobileDF \
      .where("action='open' or action='close'") \
      .withColumn("action",upper(col("action"))) \
      .select("id","action","ts") \
      .groupBy("action") \
      .count()

5）将结果DataFrame输出到控制台显示，代码如下：

# 结果输出到控制台
query = mobileDF2.writeStream \
      .format("console") \
      .option("truncate","false") \
      .outputMode("complete") \
      .start()

# 等待流程序执行结束（作为作业文件提交时启用）
# query.awaitTermination()

6）执行流处理程序，输出结果如下：

-------------------------------------------
Batch: 0
-------------------------------------------
+------+-----+
|action|count|
+------+-----+
|CLOSE |1    |
|OPEN  |3    |
+------+-----+

-------------------------------------------
Batch: 1
-------------------------------------------
+------+-----+
|action|count|
+------+-----+
|CLOSE |2    |
|OPEN  |4    |
+------+-----+

-------------------------------------------
Batch: 2
-------------------------------------------
+------+-----+
|action|count|
+------+-----+
|CLOSE |3    |
|OPEN  |5    |
+------+-----+

完整的代码如下。

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *

# 创建SparkSession实例
spark = SparkSession \
        .builder \
        .appName("streaming demo") \
        .getOrCreate()

# 设置shuffle后的分区数为10（测试环境下）
spark.conf.set("spark.sql.shuffle.partitions",10)

# 为手机事件数据创建一个schema
fields = [
      StructField("id", StringType(), nullable = False),
      StructField("action", StringType(), nullable = False),
      StructField("ts", TimestampType(), nullable = False)
]
mobileDataSchema = StructType(fields)

# 指定监听的文件目录
dataPath = "/data/spark/stream/mobile"

# 读取指定目录下的源数据文件，一次一个
mobileDF = spark.readStream \
      .option("maxFilesPerTrigger", 1) \
      .option("mode","failFast") \
      .schema(mobileDataSchema) \
      .json(dataPath)

mobileDF2 = mobileDF \
      .where("action='open' or action='close'") \
      .withColumn("action",upper(col("action"))) \
      .select("id","action","ts") \
      .groupBy("action") \
      .count()

# 结果输出到控制台
query = mobileDF2.writeStream \
      .format("console") \
      .option("truncate","false") \
      .outputMode("complete") \
      .start()

# 等待流程序执行结束（作为作业文件提交时启用）
query.awaitTermination()

在这个示例中，我们采用的输出模式是“complete”。在没有聚合操作的情况下，不能使用“complete”输出模式；在有聚合操作的情况下，不能使用“append”模式。

需要注意，在流DataFrame中，不支持以下DataFrame转换（因为它们太过复杂，无法维护状态，或者由于流数据的无界性）：

在流DataFrame上的多个聚合或聚合链。
limit和take N行。
distinct转换。
在没有任何聚合的情况下对流DataFrame进行排序。

任何使用不受支持的操作的尝试都会导致一个AnalysisException异常以及类似“XYZ操作不受流streaming DataFrame/Datasets支持”这样的消息。

小白学苑

让大数据学习更简单

结构化流DataFrame操作

选择、投影和聚合操作