Structured Streaming架构原理详解！

妖狐艹你老母 2022-12-11 12:13 251阅读 0赞

### 目录 ###

*   *  一、Structured Streaming概述
     *  二、Structured Streaming与Socket集成
     *  三、Structured Streaming与Kafka集成
     *   *  （1）通过IDEA工具
         *  （2）通过spark-shell
     *  四、Structured Streaming与MySQL集成

## 一、Structured Streaming概述 ##

Structured Streaming是一个基于sparksql引擎开发的可伸展和容错的流处理引擎。**Structured Streaming传输中的关键思想是将实时数据流视为被连续添加的表。**  
这导致了一个新的流处理模型，该模型与批处理模型非常相似。您将像在静态表上一样将流计算表示为类似于批处理的标准查询，Spark在无界输入表上将其作为增量查询运行。

**程式设计模型**  
将输入数据流视为“输入表”。流上到达的每个数据项都像是将新行附加到输入表中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
对输入的查询将生成“结果表”。在每个触发间隔（例如，每1秒钟），新行将附加到输入表中，并最终更新结果表。无论何时更新结果表，我们都希望将更改后的结果行写入外部接收器。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]  
**“输出”定义为写到外部存储器的内容。可以在不同的模式下定义输出：**

*  Complete Mode：整个更新的结果表将被写入外部存储器。由存储连接器决定如何处理整个表的写入。
 *  Append Mode：仅将自上次触发以来追加在结果表中的新行写入外部存储器。这仅适用于预期结果表中现有行不会更改的查询。
 *  Update Mode：仅自上次触发以来在结果表中已更新的行将被写入外部存储（自Spark 2.1.1起可用）。请注意，这与完成模式的不同之处在于此模式仅输出自上次触发以来已更改的行。如果查询不包含聚合，则等同于追加模式。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]

## 二、Structured Streaming与Socket集成 ##

以complete输出为例

[caizhengjie@node1 kafka]$ nc -lk 9999
    java java java 
    python java 
    java java python java hive hove
    hbase
    hbase hbase hive python java

package com.spark.test
    
    import org.apache.spark.sql.SparkSession
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/9/29 * @time : 3:36 下午 */
    object StructuredStreamingTest { 
        def main(args: Array[String]): Unit = { 
            // 创建Spark Session对象
            val spark = SparkSession
                    .builder
                    .master("local[2]")
                    .appName("HdfsTest")
                    .getOrCreate()
    
            // 创建一个流数据框架
            val lines = spark.readStream
                    .format("socket")
                    .option("host","node1")
                    .option("port",9999)
                    .load() // 返回的是dataframe格式
    
            import spark.implicits._
            // 先将dataframe准换成dataset，在对数据进行处理
            val words = lines.as[String].flatMap(_.split(" "))
            val wordCounts = words.groupBy("value").count()
    
            // 输出
            val query = wordCounts.writeStream
                    .outputMode("complete")
                    .format("console")
                    .start()
    
            query.awaitTermination()
        }
    
    }

运行结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]

## 三、Structured Streaming与Kafka集成 ##

Kafka 0.10的结构化流集成，可从Kafka读取数据或向Kafka写入数据。

### （1）通过IDEA工具 ###

对于使用Maven项目定义的Scala 应用程序，需要加载pom.xml配置文件

<dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId>
          <version>${saprk.version}</version>
    </dependency>

启动kafka的生产者

bin/kafka-console-producer.sh --broker-list node192,node2:9092,node3:9092, --topic test

>java java python python
    >hive hive java java

运行程序

package com.spark.test
    
    import org.apache.spark.sql.SparkSession
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/9/29 * @time : 4:49 下午 */
    object StructuredStreamingKafka { 
        def main(args: Array[String]): Unit = { 
            // 创建Spark Session对象
            val spark = SparkSession
                    .builder
                    .master("local[2]")
                    .appName("HdfsTest")
                    .getOrCreate()
    
            // 读取kafka stream流
            val df = spark
                    .readStream
                    .format("kafka")
                    .option("kafka.bootstrap.servers", "node1:9092")
                    .option("subscribe", "test")
                    .load() //返回的是dataframe格式
    
            import spark.implicits._
            // 先将dataframe准换成dataset，在对数据进行处理
            val lines = df.selectExpr("CAST(value AS STRING)").as[String]
            val words =  lines.flatMap(_.split(" "))
            val wordCounts = words.groupBy("value").count()
    
    
            // 输出
            val query = wordCounts.writeStream
                    .outputMode("complete")
                    .format("console")
                    .start()
    
            query.awaitTermination()
    
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]

### （2）通过spark-shell ###

在通过spark-shell运行时，需要将下面的jar包拷贝到spark的jar目录下  
![在这里插入图片描述][20201001103255381.png_pic_center]  
kafka\_2.11-2.1.1.jar和kafka-clients-2.1.1.jar在kafka的lib中能找到  
spark-sql-kafka-0-10\_2.11-2.4.6.jar和spark-streaming-kafka-0-10\_2.11-2.4.6.jar需要到maven的仓库下找  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]  
启动kafka的生产者

bin/kafka-console-producer.sh --broker-list node192,node2:9092,node3:9092, --topic test

运行spark-shell

bin/spark-shell --master local[2]

scala> :paste
    // Entering paste mode (ctrl-D to finish)
    
    val df = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "node1:9092")
    .option("subscribe", "test")
    .load()
    
    import spark.implicits._
    val lines = df.selectExpr("CAST(value AS STRING)").as[String]
    val words =  lines.flatMap(_.split(" "))
    val wordCounts = words.groupBy("value").count()
    
    
    // 输出
    val query = wordCounts.writeStream
    .outputMode("update")
    .format("console")
    .start()
    
    query.awaitTermination()
    
    
    // Exiting paste mode, now interpreting.

查看运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]  
详细介绍可以见官方文档：  
[http://spark.apache.org/docs/2.4.6/structured-streaming-kafka-integration.html][http_spark.apache.org_docs_2.4.6_structured-streaming-kafka-integration.html]

## 四、Structured Streaming与MySQL集成 ##

关于Structured Streaming与MySQL集成可以见文档：  
[https://databricks.com/blog/2017/04/04/real-time-end-to-end-integration-with-apache-kafka-in-apache-sparks-structured-streaming.html][https_databricks.com_blog_2017_04_04_real-time-end-to-end-integration-with-apache-kafka-in-apache-sparks-structured-streaming.html]

通常，我们希望能够将流的输出写入外部数据库（例如MySQL）。在撰写本文时，结构化流API不支持将外部数据库作为接收器。但是，这样做的话，API选项将像一样简单.format(“jdbc”).start(“jdbc:mysql/…”)。同时，我们可以使用foreach接收器来完成此任务。让我们创建一个自定义JDBC Sink，它扩展了ForeachWriter并实现了其方法。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]  
我们现在就可以使用我们的JDBCSink了：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]  
As batches are complete, counts by zip could be INSERTed/UPSERTed into MySQL as needed.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/a8ef38975cce4bdc837e86a92fcbdc69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/d1e3c0a0922440de9f4f4933d919477b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/d06fe5c82c4a44ef9dc458c1736dbcb9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/faca7114107741ce95f4a35bb67423e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/fa8b1177eaab40e28290459c6abbf235.png
[20201001103255381.png_pic_center]: /images/20221123/a7507261997147e5b7e7bc01c125868f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/1a16b9a2999a43bcb0cd17401a2f9cd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/2a8bb651e17248d9b7e7c58e14411559.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/b138edd97c424e448aa1c94052a37db7.png
[http_spark.apache.org_docs_2.4.6_structured-streaming-kafka-integration.html]: http://spark.apache.org/docs/2.4.6/structured-streaming-kafka-integration.html
[https_databricks.com_blog_2017_04_04_real-time-end-to-end-integration-with-apache-kafka-in-apache-sparks-structured-streaming.html]: https://databricks.com/blog/2017/04/04/real-time-end-to-end-integration-with-apache-kafka-in-apache-sparks-structured-streaming.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/d9ff6d4dcc2e46edbdea0347b82a5541.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/24cb16d10fa7417b80cb086a737cea93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/3a0bde89ba6b4189acc6a4969bbd116a.png