Spark Structured Streaming实战

Posted zackstang

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark Structured Streaming实战相关的知识,希望对你有一定的参考价值。

5. 实战Structured Streaming

5.1. Static版本

先读一份static 数据:

val static = spark.read.json("s3://tang-spark/data/activity-data/")

static.printSchema

root

 |-- Arrival_Time: long (nullable = true)

 |-- Creation_Time: long (nullable = true)

 |-- Device: string (nullable = true)

 |-- Index: long (nullable = true)

 |-- Model: string (nullable = true)

 |-- User: string (nullable = true)

 |-- gt: string (nullable = true)

 |-- x: double (nullable = true)

 |-- y: double (nullable = true)

 |-- z: double (nullable = true)

 

看一下部分内容:

技术图片

 

5.2. Streaming 版本

下面我们使用同样的数据集创建一个streaming 版本,在流的情况下,它会一个接一个地读取每个输入文件。

Streaming DataFrames与static DataFrame基本一致,基本上,所有static DF中的transformation都可以应用在Streaming DF中。不过,一个小小的区别是:Structured Streaming并不允许我们执行schema 推断(inference),除非显式地启用此功能。可以通过设置spark.sql.streaming.schemaInference 为 true来开启schema inference。在这个例子中,我们会读取一个文件的schema(当然我们已知了这些文件具有有效的schema),并从static DF中pass一个dataSchema对象到streaming DF中。不过在实际中一般要避免这么做,因为数据可能是会改变的。

val dataSchema = static.schema

val streaming = spark.readStream.schema(dataSchema).option("maxFilesPerTrigger", 1).json("s3://tang-spark/data/activity-data")

 

(这里我们指定了maxFilesPerTrigger为1,仅是为了测试方便,让stream一次读一个文件,但是一般在生产环境中不会配置这么低。)

与其他Spark API 一样,streaming DataFrame 的创建以及执行是lazy的。现在我们可以为streaming DF指定一个transformation,最后调用一个action开始执行这个流。在这个例子中,我们会展示一个简单的transformation —— 根据字段gt,进行group 并 count:

val activityCounts = streaming.groupBy("gt").count()

 

若是使用的单机模式,也可以设置一个更小的spark.sql.shuffle.partitions 值(默认为200,Configures the number of partitions to use when shuffling data for joins or aggregations.),以避免创建过多的shuffle partitions:

spark.conf.set("spark.sql.shuffle.partitions", 5)

 

现在我们已经设置好了transformation,下一步仅需要指定一个action就可以开始执行query了。根据上文所述,我们还需要指定一个输出的destination,或是针对这个query结果的output sink。对于这个简单的例子,我们会使用memory sink,它会维护一个在内存中的一个table of the results。

在指定sink时,我们还需要定义Spark如何输出它的数据,在这个例子中,我们会使用complete 的输出模式。这个模式会在每次trigger后,重写所有的keys以及它们的counts值:

val activityQuery = activityCounts.writeStream.queryName("activity_counts").format("memory").outputMode("complete").start()

 

在我们运行上面的代码前,我们也要加上这条代码:

activityQuery.awaitTermination()

 

在这条代码开始运行后,streaming 计算将会在后台运行。而这个query object即是active streaming query 对应的handle,而我们必须使用activityQuery.awaitTermination() 等待query的结束,以防止driver进程在query仍是active时退出。在生产环境中必须要加上这条语句,否则stream无法执行。

 

开始执行:

[Stage 8:==========================>                             (94 + 4) / 200]

 

可以看到有 200 个task,因为使用的是默认的 spark.sql.shuffle.partitions 值。

 

可以使用 spark.streams.active 列出active的stream:

> spark.streams.active

res24: Array[org.apache.spark.sql.streaming.StreamingQuery] = Array(org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@612c6083)

 

它也会为每个stream分配一个UUID,所以如果有必要的话,可以迭代这个list,并选出上面的那个流。不过在这个例子中,我们已经将这个流assign给一个变量了,所以没有必要。

现在这个流正在运行,我们可以通过query这个in-memory table来检查结果,这个表维护了当前streaming aggregation的输出。这个table将被称为 activity_counts,与流的名称一样。在检查当前output table 的输出时,仅需要query此表即可。我们会写一个循环,每秒打出这个streaming query 的结果:

for (i <- 1 to 5){

     spark.sql("select * from activity_counts").show()

     Thread.sleep(1000)

   }

 

结果类似于:

技术图片

 

整体语句为:

val static = spark.read.json("s3://tang-spark/data/activity-data/")

 

val dataSchema = static.schema

val streaming = spark.readStream.schema(dataSchema).option("maxFilesPerTrigger", 1).json("s3://tang-spark/data/activity-data")

 

val activityCounts = streaming.groupBy("gt").count()

val activityQuery = activityCounts.writeStream.queryName("activity_counts").format("memory").outputMode("complete").start()

 

for (i <- 1 to 5){

     spark.sql("select * from activity_counts").show()

     Thread.sleep(1000)

   }

 

6. Streams 中的 Transformations

Streaming中的transformations 基本涵盖了static DF 中的 transformations,例如select、filter等简单的transformations。不过仍是有些限制,例如在Spark 2.2 版本中,用户无法sort一个没有aggregated的stream,并且在没有使用Stateful Processing时,无法执行多层的aggregation,等等。这些限制可能会随着版本更替有所解决,对此最好参考Spark的官方文档,查看更新情况。

 

6.1. Selections and Filtering

在Structured Streaming中,支持所有的select 与 filter。下面是一个简单的例子,由于我们并不会随着时间更新key,所以会使用Append output mode,将新数据append到输出表中:

import org.apache.spark.sql.functions.expr

val simpleTransform = streaming.withColumn("stairs", expr("gt like ‘%stairs%‘"))

  .where("stairs")

  .where("gt is not null")

  .select("gt", "model", "arrival_time", "creation_time")

  .writeStream

  .queryName("simple_transform")

  .format("memory")

  .outputMode("append")

  .start()

 

下面查询一下,先可以看看有哪些表:

技术图片

 

然后查询 simple_transform 表:

spark.sql("select * from simple_transform").show()

技术图片

 

6.2. Aggregations

Structured Streaming 对 aggregation操作的支持非常好。我们可以指定任意聚合操作。例如,我们可以使用一个比较有特点的聚合(例如Cube),在sensor的phone model 、activity、以及average x,y,z acceleration 字段上聚合Cube:

val deviceModelStats = streaming.cube("gt", "model").avg()

  .drop("avg(Arrival_time)")

  .drop("avg(Creation_Time)")

  .drop("avg(Index)")

  .writeStream.queryName("device_counts").format("memory").outputMode("complete")

  .start()

 

spark.sql("select * from device_counts").show()

技术图片

 

相信大家已经发现了,这里queryName(“device_counts”) 也就是我们最后生成的表名。

 

6.3. Joins

在Spark 2.2 版本以后,Structured Streaming支持streaming DataFrames 与 static DataFrames进行join。Spark 2.3 会增加多个streams 的join 功能。

val historicalAgg = static.groupBy("gt", "model").avg()

val deviceModelStats = streaming.drop("Arrival_Time", "Creation_Time", "Index")

  .cube("gt", "model").avg()

  .join(historicalAgg, Seq("gt", "model"))

  .writeStream.queryName("device_counts_join").format("memory").outputMode("complete")

  .start()

 

spark.sql("select * from device_counts_join").show()

技术图片

 

在Spark 2.2 中,full outer join、left join(stream在右)、right join(stream在左)暂不支持。Structured Streaming 也还没有支持stream-to-stream 的join,不过这个也是一个正在开发中的功能。

 

7. Input 与 Output

这节我们会深入讨论,在Structured Streaming 中,sources、sinks以及output modes 是如何工作的。特别地,我们会讨论数据流是怎样、何时、从哪,进入以及流出系统。

下面讨论的source 与 sink 都是端到端相同的情况一起讨论的,当然端到端也可以是不一样的(例如source 是kafka,sink 是file)

 

7.1. 数据在哪读写(Sources 与 Sinks)

Structured Streaming 支持多个production sources 与 sinks(files 以及Kafka),以及一些debug 工具(例如memory table sink)。我们之前已经介绍过这些,现在进一步讨论。

 

File Source 与 Sink

可能能想到的最简单的source就是file source,基本上我们能想到的所有文件类型都适用,如Parquet、text、JSON、CSV等。

在Spark static file source 与Structured Streaming file source 中,它们唯一的区别就是:在streaming 中,我们可以控制每次trigger中读入的文件数,通过maxFilesPerTriger选项即可。

需要注意的是,任何添加到Streaming input 目录中的文件,都必须是原子的。否则,在你写入完成前,Spark会partially 处理文件。在如本地文件系统或是HDFS这种允许partial write 文件系统中,最好是在外部文件夹写完文件后,再移动到 input 目录中。在Amazon S3 中,对象一般仅在fully writes 后才可被访问。

 

Kafka Source 与 Sink

Kafka 是一个分布式的publish-and-subscribe 系统,它可以像一个消息队列一样,让我们可以publish records 到流中。可以将Kafka 认作为一个分布式的buffer,它让我们在某个类别中(topic)存储流中的records。

 

读Kafka Source

在读kafka 时,需要选择一种选项:assign,subscribe,或是subscribePattern。在读Kafka时,只能选择其中一种选项。

  • Assign :是一种细粒度的配置模式,不仅需要指定topic,还需要指定你想要读的topic 的partitions。这个通过一个JSON string来配置,例如{“topicA”: [0, 1], “topicB”: [2, 4]}。
  • Subscribe与SubscribePattern:是用于subscribe 一个或多个topics 的模式,可以指定一个list(subscribe),或是一个pattern(subscribePattern)

然后,我们还需要指定kafka.bootstrap.servers,除此之外,还有其他几个options需要指定:

  • StartingOffsets 与 endingOffsets:query开始时的初始点,可以是earliest,也就是最开始的offsets;latest,也就是最新的offsets;或者是一个JSON string,为每个TopicPartition指定的初始offset。在JSON中,-2 作为offset可以用于指向earliest,-1指向latest。例如,一个JSON定义为 {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}。
  • failOnDataLoss:如果有数据丢失(例如topics 被删除、或者offsets 超出范围),是否要fail掉query。默认是true,可以根据任务需求进行修改。
  • maxOffsetsPerTrigger:对于给定的一个trigger,读入多少个offsets

当然,还有其他配置选项,如Kafka consumer timeouts,fetch retries,以及 intervals 等。

下面是一个例子:

// subscribe to 1 topic

val ds1 = spark.readStream.format("kafka")

  .option("kafka.bootstrap.servers", "b-1.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092,b-2.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092")

  .option("subscribe", "topic1")

  .load()

 

// subscribe to multiple topics

val ds1 = spark.readStream.format("kafka")

  .option("kafka.bootstrap.servers", "b-1.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092,b-2.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092")

  .option("subscribe", "topic1, topic2")

  .load()

 

// subscribe to a pattern of topics

val ds1 = spark.readStream.format("kafka")

  .option("kafka.bootstrap.servers", "b-1.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092,b-2.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092")

  .option("subscribePattern", "topic.*")

  .load()

 

这里若是弹出以下报错:

org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".

 

则根据Spark 官网添加jar 包:

spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.5

或:

./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.5 ...

 

在Kafka Source 中的每条row都会有以下schema:

  • key: binary
  • value: binary
  • topic: string
  • partition: int
  • offset: long
  • timestamp: long

 

在Kafka中的每条message都会以某种方式序列化。可以使用Spark原生的方法或是UDF将message 处理为一个更具结构化的形式,常见的模式有JSON或是Avro,去读写Kafka。

 

写Kafka Sink

写 Kafka Sink与读Kafka Sink 类似,仅有几个参数不一样。我们仍需要指定Kafka bootstrap servers,而其他需要提供的选项是topic以及列名,有两种方式提供,如下所示,下面两个写法是等价的:

// specify topic and columns together

ds1.selectExpr("topic", "CAST(key as STRING)", "CAST(value AS STRING)")

  .writeStream.format("kafka")

  .option("checkpointLocation", "/kafka/checkpoint/dir")

  .option("kafka.bootstrap.servers", "b-1.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092,b-2.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092")

  .start()

 

// specify column and topic seperately

ds1.selectExpr("CAST(key as STRING)", "CAST(value AS STRING)")

  .writeStream.format("kafka")

  .option("checkpointLocation", "/kafka/checkpoint/dir")

  .option("kafka.bootstrap.servers", "b-1.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092,b-2.tang-kafka-plain.hnjunx.c4.kafka.cn-north-1.amazonaws.com.cn:9092")

  .option("topic", "topic1")

  .start()

 

foreach Sink

foreach sink 类似于Dataset API 中的 foreachPartitions。这个操作允许operations可以在per-partition 的基础下进行并行地计算。在使用foreach sink时,必须实现ForeachWriter 接口,可以在Scala/Java 文档中查看,它包含3个方法:open、process 和 close。在trigger触发了一组生成的rows后,会执行相对应的这三个方法。

总的来说,这个是一个自定义的方法,可以实现自己想要的业务逻辑。

 

针对测试的Sources与Sinks

这些主要用于做测试,一般不会在生产环境中使用,因为它们无法提供端到端的fault tolerance。

 

Socket source

通过TCP sockets发送的数据,指定host和port 来读数据,如:

val socketDF = spark.readStream.format("socket")

  .option("host", "localhost").option("port", 9999).load()

 

在客户端可以执行:

nc -lk 9999

 

Console sink

可以将输出吸入到console,支持append和complete的输出模式:

activityCounts.format("console").write()

 

Memory sink

类似于console sink,区别是会将数据collect到driver中,然后以in-memory table的形式提供交互式query:

activityCounts.writeStream.format("memory").queryName("my_device_table")

 

7.2. 数据如何输出(Output Modes)

在Structured Streaming 中,output modes的概念与static DF 中的一样。

 

Append Mode

Append mode 是默认的行为,也是最简单最容易理解的。在一个新row添加到result table时,它们会基于指定的trigger,输出到sink。在fault-tolerant sink的条件下,这个模式可以确保每行row仅输出一次(并且仅此一次)。当我们将append mode 与event-time 和watermarks 结合使用时,仅有最终的结果会输出到sink。

 

Complete Mode

Complete mode 会输出result table 的整个state到配置的output sink 中。在一些stateful 数据的场景下,例如所有的rows都会随时间改变,或是配置的sink 并不支持行级的updates 时,complete mode 非常有用。如果query 并不包含aggregation,则这个等同于 append mode

 

Update Mode

Update mode 类似于 complete mode,区别是:仅有与上一次不同的rows 会被写入到sink 中。不过,sink必须支持行级的 updates,才能支持这个模式。如果query 不包含任何 aggregation,则这个模式等同于 append mode。

 

什么场景选择各个mode?

Structured Streaming 对于每个output mode,都会结合query的信息进行限制。例如,假设我们的query仅是做一个map操作,那么Structured Streaming不会允许complete mode,因为这样会需要记录所有的输入records,并重写整个output table。下面是一些小手册,关于什么时候使用各个output mode。

 技术图片

什么时候数据输出(Triggers)

在控制什么时候将数据输出到sink时,我们会设置一个trigger。默认情况下,Structured Streaming会在上一个trigger完成处理后,启动数据。我们可以使用triggers来确保不会有太多的updates压垮output sink,或是控制output中的文件大小。当前,仅有一种定期触发的trigger类型,基于的是processing time;以及一次性的trigger,用于手动一次执行一个processing step。在未来会加入更多的triggers。

 

Processing Time Trigger

对于processing time trigger,我们仅需要指定一个string 类型的duration时间即可(或是在Scalat中的Duration 类型、Java中的TimeUnit类型),例如:

import org.apache.spark.sql.streaming.Trigger

activityCounts.writeStream

  .trigger(Trigger.ProcessingTime("100 seconds"))

  .format("memory")

  .outputMode("complete")

  .start()

 

ProcessingTime trigger 会等到给定duration的时间再输出数据。例如,假设给定的duration是1分钟,则trigger会在12:00,12:01,12:02 触发,依次类推。如果由于上一个处理没完成,导致了一个trigger time miss,则Spark会等到下一个trigger 点(也就是下一分钟),而不是在上一个处理完成后立即终止。

 

Once Trigger

我们也可以仅让一个streaming job 执行一次,指定once trigger即可。这个听起来有点奇怪,不过在开发与生产中,这个trigger都非常有用。对于开发,可以用于测试。而对于生产,Once trigger可以让我们以一个非常低频的方式执行任务(例如,向数据源导入数据是一个不确定的时间周期)。由于Structured Streaming 仍可以追溯所有已经处理过的文件,以及计算后的state,它比自身再写一个追溯batch job 的处理要简单地多,并且可以节省较多的资源(因为流程序定期运行的话,会一直消耗计算资源)。使用方式如下:

  import org.apache.spark.sql.streaming.Trigger

 

activityCounts.writeStream.trigger(Trigger.Once())

  .format("console")

  .outputMode("complete")

  .start()

 

8. Streaming Dataset API

最后一点需要提到的是,在Structured Streaming 中,不仅只限于对stream使用DataFrame API。还可以使用Datasets执行同样的计算,但是用的是type-safe 的方式。我们可以将一个streaming DataFrame转换为一个Dataset。

 

以上是关于Spark Structured Streaming实战的主要内容,如果未能解决你的问题,请参考以下文章

Spark Structured Streaming

Spark Structured Streaming

Spark Structured Streaming - 1

删除由 spark-structured-streaming 写入的损坏的 parquet 文件时,我会丢失数据吗?

无法使用Spark Structured Streaming在Parquet文件中写入数据

如何使用Spark Structured Streaming连续监视目录