Structured Streaming Source

使用 Structured Streaming 最重要的就是对 Streaming DataFrame 和 Streaming DataSet 进行各种操作.

从 Spark2.0 开始, DataFrame 和 DataSet 可以表示静态有界的表, 也可以表示流式无界表.

与静态 Datasets/DataFrames 类似，我们可以使用公共入口点 SparkSession 从流数据源创建流式 Datasets/DataFrames，并对它们应用与静态 Datasets/DataFrames 相同的操作。

通过spark.readStream()得到一个DataStreamReader对象, 然后通过这个对象加载流式数据源, 就得到一个流式的 DataFrame.

park 内置了几个流式数据源, 基本可以满足我们的所有需求.

1. File source 读取文件夹中的文件作为流式数据. 支持的文件格式: text, csv, josn, orc, parquet. 注意, 文件必须放置的给定的目录中, 在大多数文件系统中, 可以通过移动操作来完成.
2. kafka source 从 kafka 读取数据. 目前兼容 kafka 0.10.0+ 版本
3. socket source 用于测试. 可以从 socket 连接中读取 UTF8 的文本数据. 侦听的 socket 位于驱动中. 注意, 这个数据源仅仅用于测试. 做不到exactly-once，因为socket停了数据就没了
4. rate source 用于测试. 以每秒指定的行数生成数据，每个输出行包含一个 timestamp 和 value。其中 timestamp 是一个 Timestamp类型(信息产生的时间)，并且 value 是 Long 包含消息的数量. 用于测试和基准测试.

socket source

具体案例参考前面的快速入门

file source

路径指定到文件夹 ，readStream必须指定schema；read可以推断文件类型

读取普通文件夹内的文件

• ```scala

  import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
  import org.apache.spark.sql.types.{LongType, StringType, StructType}
  import org.apache.spark.sql.{DataFrame, SparkSession}

  object ReadFromFile {
  def main(args: Array[String]): Unit = {
  val spark: SparkSession = SparkSession
  .builder()
  .master(“local[*]”)
  .appName(“ReadFromFile”)
  .getOrCreate()

      // 定义 Schema, 用于指定列名以及列中的数据类型
      val userSchema: StructType = new StructType().add("name", StringType).add("age", LongType).add("job", StringType)
      val user: DataFrame = spark.readStream
          .format("csv")
      
          .schema(userSchema) // 指定schema
      
          .load("/Users/lzc/Desktop/csv")  // 必须是目录, 不能是文件名
  
      val query: StreamingQuery = user.writeStream
          .outputMode("append")
          .trigger(Trigger.ProcessingTime(0)) // 触发器 数字表示毫秒值. 0 表示立即处理
          .format("console")
          .start()
      query.awaitTermination()
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

  }

注意: 一个文件读取完 ，更改文件内数据是监听不到

前面获取user的代码也可以使用下面的替换:

val user: DataFrame = spark.readStream
            .schema(userSchema)
            .csv("/Users/lzc/Desktop/csv")
1
2
3

读取自动分区的文件夹内的文件

当文件夹被命名为 “key=value” 形式时, Structured Streaming 会自动递归遍历当前文件夹下的所有子文件夹, 并根据文件名实现自动分区.

如果文件夹的命名规则不是"key=value"形式, 则不会触发自动分区. 另外, 同级目录下的文件夹的命名规则必须一致.

• 步骤 1: 创建如下文件内容:

user1.csv

lisi,male,18
zhiling,female,28
1
2

user2.csv

lili,femal,19
fengjie,female,40
1
2

• 步骤 2: 创建如下代码

import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

object ReadFromFile2 {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("ReadFromFile")
            .getOrCreate()

        // 定义 Schema, 用于指定列名以及列中的数据类型
        val userSchema: StructType = new StructType().add("name", StringType).add("sex", StringType).add("age", IntegerType)

        val user: DataFrame = spark.readStream
            .schema(userSchema)
            .csv("/Users/lzc/Desktop/csv")

        val query: StreamingQuery = user.writeStream
            .outputMode("append")
            .trigger(Trigger.ProcessingTime(0)) // 触发器 数字表示毫秒值. 0 表示立即处理
            .format("console")
            .start()
        query.awaitTermination()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Kafka source

参考文档: http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html

以 Streaming 模式创建 Kafka 工作流

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{DataFrame, SparkSession}

object KafkaSourceDemo {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("KafkaSourceDemo")
            .getOrCreate()

        // 得到的 df 的 schema 是固定的: key,value,topic,partition,offset,timestamp,timestampType
        val df: DataFrame = spark.readStream
            .format("kafka") // 设置 kafka 数据源
            .option("kafka.bootstrap.servers", "hadoop201:9092,hadoop202:9092,hadoop203:9092")
            .option("subscribe", "topic1") // 也可以订阅多个主题:   "topic1,topic2"
            .load


        df.writeStream
            .outputMode("update")
            .format("console")
            .trigger(Trigger.Continuous(1000))
            // timestamp显示完整时间
            .option("truncate",false)
            .start
            .awaitTermination()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

对于kafka来说格式是固定的：

value：字节数组

timestamp：kafka写入数据时间戳

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object KafkaSourceDemo2 {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("KafkaSourceDemo")
            .getOrCreate()
        import spark.implicits._
        // 得到的 df 的 schema 是固定的: key,value,topic,partition,offset,timestamp,timestampType
        val lines: Dataset[String] = spark.readStream
            .format("kafka") // 设置 kafka 数据源
            .option("kafka.bootstrap.servers", "hadoop201:9092,hadoop202:9092,hadoop203:9092")
            .option("subscribe", "topic1") // 也可以订阅多个主题:   "topic1,topic2"
            .load
            // 转换成string
            .selectExpr("CAST(value AS string)")
            .as[String]
        val query: DataFrame = lines.flatMap(_.split("\\W+")).groupBy("value").count()
        query.writeStream
            .outputMode("complete")
            .format("console")
            .option("checkpointLocation", "./ck1")  // 下次启动的时候, 可以从上次的位置开始读取
            .start
            .awaitTermination()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

通过 Batch 模式创建 Kafka 工作流

这种模式一般需要设置消费的其实偏移量和结束偏移量, 如果不设置 checkpoint 的情况下, 默认起始偏移量 earliest, 结束偏移量为 latest.

该模式为一次性作业(批处理), 而非持续性的处理数据.

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object KafkaSourceDemo3 {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("KafkaSourceDemo")
            .getOrCreate()
        import spark.implicits._

        val lines: Dataset[String] = spark.read  // 使用 read 方法,而不是 readStream 方法
            .format("kafka") // 设置 kafka 数据源
            .option("kafka.bootstrap.servers", "hadoop201:9092,hadoop202:9092,hadoop203:9092")
            .option("subscribe", "topic1")
            .option("startingOffsets", "earliest")
            .option("endingOffsets", "latest")
            .load
            .selectExpr("CAST(value AS STRING)")
            .as[String]

        val query: DataFrame = lines.flatMap(_.split("\\W+")).groupBy("value").count()

        query.write   // 使用 write 而不是 writeStream
            .format("console")
            .save()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

Rate Source

以固定的速率生成固定格式的数据, 用来测试 Structured Streaming 的性能.

import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{DataFrame, SparkSession}

object RateSourceDemo {
    def main(args: Array[String]): Unit = {
        val spark: SparkSession = SparkSession
            .builder()
            .master("local[*]")
            .appName("RateSourceDemo")
            .getOrCreate()

        val rows: DataFrame = spark.readStream
            .format("rate") // 设置数据源为 rate
            .option("rowsPerSecond", 10) // 设置每秒产生的数据的条数, 默认是 1
            .option("rampUpTime", 1) // 设置多少秒到达指定速率 默认为 0
            .option("numPartitions", 2) /// 设置分区数  默认是 spark 的默认并行度
            .load

        rows.writeStream
            .outputMode("append")
            .trigger(Trigger.Continuous(1000))
            .format("console")
            .start()
            .awaitTermination()
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27