结构化流DataFrame操作

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

选择、投影和聚合操作

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

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

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

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）先导入相关的依赖包。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

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

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

    // 为手机事件数据创建一个schema
    val fields = Array(
      StructField("id", StringType, nullable = false),
      StructField("action", StringType, nullable = false),
      StructField("ts", TimestampType, nullable = false)
    )
    val mobileDataSchema = StructType(fields)

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

    // 监听的文件目录
    val dataPath = "src/main/data/mobile"

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

4）执行过滤、投影、聚合等转换操作。

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

5）将结果DataFrame输出到控制台显示。

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

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    |
+------+-----+

完整的代码如下。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
......

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local")
      .appName("file source")
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")      // 设置日志级别

    // 为手机事件数据创建一个schema
    val fields = Array(
      StructField("id", StringType, nullable = false),
      StructField("action", StringType, nullable = false),
      StructField("ts", TimestampType, nullable = false)
    )
    val mobileDataSchema = StructType(fields)

    // 监听的文件目录
    val dataPath = "src/main/data/mobile"

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

    // 选择、投影、聚合等操作
    val mobileDF2 = mobileDF
      .where("action='open' or action='close'")
      .withColumn("action",upper(col("action")))
      .select("id","action","ts")
      .groupBy("action")
      .count

    // 也可以创建一个视图来应用SQL查询
    // mobileDF2.createOrReplaceTempView("clean_mobile")
    // val sqlDF = spark.sql("select id,action,ts from clean_mobile")
    // val sqlDF = spark.sql("select action,count(*) as cnt from clean_mobile group by action")

    // 结果输出到控制台（注意，输出模式设置）
    val 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操作

选择、投影和聚合操作