输出终端/位置

文件接收器

Memory Sink

Foreach和ForeachBatch Sink

Foreach

ForeachBatch

代码演示

输出终端/位置

Structured Streaming 非常显式地提出了输入(Source)、执行(StreamExecution)、输出(Sink)的3个组件，并且在每个组件显式地做到fault-tolerant（容错），由此得到整个streaming程序的 end-to-end exactly-once guarantees。

目前Structured Streaming内置FileSink、Console Sink、Foreach Sink（ForeachBatch Sink）、Memory Sink及Kafka Sink，其中测试最为方便的是Console Sink。

文件接收器

将输出存储到目录文件中，支持文件格式：parquet、orc、json、csv等，示例如下：

Memory Sink

此种接收器作为调试使用，输出作为内存表存储在内存中，支持Append和Complete输出模式。这应该用于低数据量的调试目的，因为整个输出被收集并存储在驱动程序的内存中，因此，请谨慎使用，示例如下：

Foreach和ForeachBatch Sink

Foreach

Structured Streaming提供接口foreach和foreachBatch，允许用户在流式查询的输出上应用任意操作和编写逻辑，比如输出到MySQL表、Redis数据库等外部存系统。其中foreach允许每行自定义写入逻辑，foreachBatch允许在每个微批量的输出上进行任意操作和自定义逻辑，建议使用foreachBatch操作。

foreach表达自定义编写器逻辑具体来说，需要编写类class继承ForeachWriter，其中包含三个方法来表达数据写入逻辑：打开，处理和关闭。

https://databricks.com/blog/2017/04/04/real-time-end-to-end-integration-with-apache-kafka-in-apache-sparks-structured-streaming.html

streamingDatasetOfString.writeStream.foreach(new ForeachWriter[String] {def open(partitionId: Long, version: Long): Boolean = {// Open connection}def process(record: String): Unit = {// Write string to connection}def close(errorOrNull: Throwable): Unit = {// Close the connection}}).start()

ForeachBatch

方法foreachBatch允许指定在流式查询的每个微批次的输出数据上执行的函数，需要两个参数：微批次的输出数据DataFrame或Dataset、微批次的唯一ID。

使用foreachBatch函数输出时，以下几个注意事项：

1.重用现有的批处理数据源，可以在每个微批次的输出上使用批处理数据输出Output；

2.写入多个位置，如果要将流式查询的输出写入多个位置，则可以简单地多次写入输出 DataFrame/Dataset 。但是，每次写入尝试都会导致重新计算输出数据（包括可能重新读取输入数据）。要避免重新计算，您应该缓存cache输出 DataFrame/Dataset，将其写入多个位置，然后 uncache 。

3.应用其他DataFrame操作，流式DataFrame中不支持许多DataFrame和Dataset操作，使用foreachBatch可以在每个微批输出上应用其中一些操作，但是，必须自己解释执行该操作的端到端语义。

4.默认情况下，foreachBatch仅提供至少一次写保证。但是，可以使用提供给该函数的batchId作为重复数据删除输出并获得一次性保证的方法。

5.foreachBatch不适用于连续处理模式，因为它从根本上依赖于流式查询的微批量执行。如果以连续模式写入数据，请改用foreach。

代码演示

使用foreachBatch将词频统计结果输出到MySQL表中，代码如下：

package cn.itcast.structedstreamingimport org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}/*** 使用Structured Streaming从TCP Socket实时读取数据，进行词频统计，将结果存储到MySQL数据库表中*/
object StructuredForeachBatch {def main(args: Array[String]): Unit = {val spark: SparkSession = SparkSession.builder().appName(this.getClass.getSimpleName.stripSuffix("$")).master("local[*]").config("spark.sql.shuffle.partitions", "2").getOrCreate()val sc: SparkContext = spark.sparkContextsc.setLogLevel("WARN")import spark.implicits._import org.apache.spark.sql.functions._val inputStreamDF: DataFrame = spark.readStream.format("socket").option("host", "node1").option("port", 9999).load()val resultStreamDF: DataFrame = inputStreamDF.as[String].filter(StringUtils.isNotBlank(_)).flatMap(_.trim.split("\\s+")).groupBy($"value").count()val query: StreamingQuery = resultStreamDF.writeStream.outputMode(OutputMode.Complete()).foreachBatch((batchDF: DataFrame, batchId: Long) => {println(s"BatchId = ${batchId}")if (!batchDF.isEmpty) {batchDF.coalesce(1).write.mode(SaveMode.Overwrite).format("jdbc")//.option("driver", "com.mysql.cj.jdbc.Driver")//MySQL-8//.option("url", "jdbc:mysql://localhost:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true")//MySQL-8.option("url", "jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8").option("user", "root").option("password", "root").option("dbtable", "bigdata.t_struct_words").save()}}).start()query.awaitTermination()query.stop()}
}

2021年大数据Spark（四十八）：Structured Streaming 输出终端/位置相关推荐

2021年大数据Spark（十八）：Spark Core的RDD Checkpoint
目录 RDD Checkpoint 引入 API 代码演示总结:持久化和Checkpoint的区别问题: 答案: 区别: RDD Checkpoint 引入 RDD 数据可以持久化,但是持久化/缓 ...
2021年大数据ELK（十八）：Beats 简单介绍和FileBeat工作原理
全网最详细的大数据ELK文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录 Beats 简单介绍和FileBeat工作原理一.Beats 二.FileB ...
2021年大数据Spark（十四）：Spark Core的RDD操作
目录 RDD的操作函数(算子)分类 Transformation函数 Action函数 RDD的操作有一定开发经验的读者应该都使用过多线程,利用多核 CPU 的并行能力来加快运算速率 ...
客快物流大数据项目(四十八)：Spark操作Kudu 修改表
Spark操作Kudu 修改表代码示例 /*** 添加列* @param kuduContext*/ def addColumn(kuduContext: KuduContext): Unit ={ ...
2021年大数据Spark（十二）：Spark Core的RDD详解
目录 RDD详解为什么需要RDD? 什么是RDD? RDD的5大特性第一个:A list of partitions 第二个:A function for computing each split ...
2021年大数据Spark（十九）：Spark Core的共享变量
目录共享变量广播变量累加器案例演示共享变量在默认情况下,当Spark在集群的多个不同节点的多个任务上并行运行一个函数时,它会把函数中涉及到的每个变量,在每个任务上都生成一个副 ...
2021年大数据Spark（十五）：Spark Core的RDD常用算子
目录常用算子基本算子分区操作函数算子重分区函数算子 1).增加分区函数 2).减少分区函数 3).调整分区函数聚合函数算子 Scala集合中的聚合函数 ...
2021年大数据Flink（十八）：Flink Window操作
目录 Flink-Window操作为什么需要Window Window的分类按照time和count分类按照slide和size分类总结 Window ...
2021年大数据Spark（十六）：Spark Core的RDD算子练习
目录 RDD算子练习 map 算子 filter 算子 flatMap 算子交集.并集.差集.笛卡尔积 distinct 算子 first.take.top 算子 ...
2021年大数据Hadoop（十四）：HDFS的高可用机制
全网最详细的Hadoop文章系列,强烈建议收藏加关注! 后面更新文章都会列出历史文章目录,帮助大家回顾知识重点. 目录本系列历史文章前言 HDFS的高可用机制 HDFS高可用介绍组件介绍 Nam ...

2021年大数据Spark（四十八）：Structured Streaming 输出终端/位置

输出终端/位置

文件接收器

Memory Sink

Foreach和ForeachBatch Sink

Foreach

ForeachBatch

代码演示

2021年大数据Spark（四十八）：Structured Streaming 输出终端/位置相关推荐

最新文章

热门文章

2021年大数据Spark（四十八）：Structured Streaming 输出终端/位置

输出终端/位置

​​​​​​​文件接收器

​​​​​​​Memory Sink

Foreach和ForeachBatch Sink

Foreach

​​​​​​​ForeachBatch

​​​​​​​代码演示

2021年大数据Spark（四十八）：Structured Streaming 输出终端/位置相关推荐

最新文章

热门文章

文件接收器

Memory Sink

ForeachBatch

代码演示