devbox
煮酒与君饮
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spark实时（三）：Structured Streaming入门案例_spark structured streaming kafka java示例

作者：煮酒与君饮 | 2024-07-30 14:54:23

赞

踩

spark structured streaming kafka java示例

文章目录

Structured Streaming入门案例

一、Scala代码如下

二、Java 代码如下

三、以上代码注意点如下

Structured Streaming入门案例

我们使用Structured Streaming来监控socket数据统计WordCount。这里我们使用Spark版本为3.4.3版本，首先在Maven pom文件中导入以下依赖：

  1.  <!-- 配置以下可以解决 在jdk1.8环境下打包时报错 “-source 1.5 中不支持 lambda 表达式” -->
  2.   <properties>
  3.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  4.     <maven.compiler.source>1.8</maven.compiler.source>
  5.     <maven.compiler.target>1.8</maven.compiler.target>
  6.     <spark.version>3.4.3</spark.version>
  7.   </properties>
  8.  
  9.   <dependencies>
  10.     <!-- Spark-core -->
  11.     <dependency>
  12.       <groupId>org.apache.spark</groupId>
  13.       <artifactId>spark-core_2.12</artifactId>
  14.       <version>${spark.version}</version>
  15.     </dependency>
  16.     <!-- SparkSQL -->
  17.     <dependency>
  18.       <groupId>org.apache.spark</groupId>
  19.       <artifactId>spark-sql_2.12</artifactId>
  20.       <version>${spark.version}</version>
  21.     </dependency>
  22.     <!-- SparkSQL  ON  Hive-->
  23.     <dependency>
  24.       <groupId>org.apache.spark</groupId>
  25.       <artifactId>spark-hive_2.12</artifactId>
  26.       <version>${spark.version}</version>
  27.     </dependency>
  28.     <!--mysql依赖的jar包-->
  29.     <dependency>
  30.       <groupId>mysql</groupId>
  31.       <artifactId>mysql-connector-java</artifactId>
  32.       <version>5.1.47</version>
  33.     </dependency>
  34.     <!--SparkStreaming-->
  35.     <dependency>
  36.       <groupId>org.apache.spark</groupId>
  37.       <artifactId>spark-streaming_2.12</artifactId>
  38.       <version>${spark.version}</version>
  39.     </dependency>
  40.  
  41.     <!-- Kafka 0.10+ Source For Structured Streaming-->
  42.     <dependency>
  43.       <groupId>org.apache.spark</groupId>
  44.       <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
  45.       <version>${spark.version}</version>
  46.     </dependency>
  47.  
  48.     <!-- 向kafka 生产数据需要包 -->
  49.     <dependency>
  50.       <groupId>org.apache.kafka</groupId>
  51.       <artifactId>kafka-clients</artifactId>
  52.       <version>2.8.0</version>
  53.     </dependency>
  54.  
  55.     <!-- Scala 包-->
  56.     <dependency>
  57.       <groupId>org.scala-lang</groupId>
  58.       <artifactId>scala-library</artifactId>
  59.       <version>2.12.15</version>
  60.     </dependency>
  61.     <dependency>
  62.       <groupId>org.scala-lang</groupId>
  63.       <artifactId>scala-compiler</artifactId>
  64.       <version>2.12.15</version>
  65.     </dependency>
  66.     <dependency>
  67.       <groupId>org.scala-lang</groupId>
  68.       <artifactId>scala-reflect</artifactId>
  69.       <version>2.12.15</version>
  70.     </dependency>
  71.     <dependency>
  72.       <groupId>log4j</groupId>
  73.       <artifactId>log4j</artifactId>
  74.       <version>1.2.12</version>
  75.     </dependency>
  76.     <dependency>
  77.       <groupId>com.google.collections</groupId>
  78.       <artifactId>google-collections</artifactId>
  79.       <version>1.0</version>
  80.     </dependency>
  81.  
  82.   </dependencies>

一、Scala代码如下

  1. package com.lanson.structuredStreaming
  2.  
  3. /**
  4.  *  Structured Streaming 实时读取Socket数据
  5.  */
  6.  
  7. import org.apache.spark.sql.streaming.StreamingQuery
  8. import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
  9.  
  10. /**
  11.  * Structured Streaming 读取Socket数据
  12.  */
  13. object SSReadSocketData {
  14.   def main(args: Array[String]): Unit = {
  15.  
  16.     //1.创建SparkSession对象
  17.     val spark: SparkSession = SparkSession.builder()
  18.       .master("local")
  19.       .appName("StructuredSocketWordCount")
  20.       //默认200个并行度，由于源头数据量少，可以设置少一些并行度
  21.       .config("spark.sql.shuffle.partitions",1)
  22.       .getOrCreate()
  23.  
  24.     import spark.implicits._
  25.  
  26.     spark.sparkContext.setLogLevel("Error")
  27.  
  28.     //2.读取Socket中的每行数据,生成DataFrame默认列名为"value"
  29.     val lines: DataFrame = spark.readStream
  30.       .format("socket")
  31.       .option("host", "node3")
  32.       .option("port", 9999)
  33.       .load()
  34.  
  35.     //3.将每行数据切分成单词，首先通过as[String]转换成Dataset操作
  36.     val words: Dataset[String] = lines.as[String].flatMap(line=>{line.split(" ")})
  37.  
  38.     //4.按照单词分组，统计个数，自动多一个列count
  39.     val wordCounts: DataFrame = words.groupBy("value").count()
  40.  
  41.     //5.启动流并向控制台打印结果
  42.     val query: StreamingQuery = wordCounts.writeStream
  43.       //更新模式设置为complete
  44.       .outputMode("complete")
  45.       .format("console")
  46.       .start()
  47.     query.awaitTermination()
  48.  
  49.   }
  50.  
  51. }

 

二、Java 代码如下

  1. package com.lanson.structuredStreaming;
  2.  
  3. import java.util.Arrays;
  4. import java.util.Iterator;
  5. import java.util.concurrent.TimeoutException;
  6. import org.apache.spark.api.java.function.FlatMapFunction;
  7. import org.apache.spark.sql.Dataset;
  8. import org.apache.spark.sql.Encoders;
  9. import org.apache.spark.sql.Row;
  10. import org.apache.spark.sql.SparkSession;
  11. import org.apache.spark.sql.streaming.StreamingQuery;
  12. import org.apache.spark.sql.streaming.StreamingQueryException;
  13.  
  14. public class SSReadSocketData01 {
  15.  
  16.     public static void main(String[] args) throws StreamingQueryException, TimeoutException {
  17.         SparkSession spark = SparkSession.builder().master("local")
  18.             .appName("SSReadSocketData01")
  19.             .config("spark.sql.shuffle.partitions", 1)
  20.             .getOrCreate();
  21.         
  22.         spark.sparkContext().setLogLevel("Error");
  23.  
  24.         Dataset<Row> lines = spark.readStream().format("socket")
  25.             .option("host", "node3")
  26.             .option("port", 9999)
  27.             .load();
  28.  
  29.         Dataset<String> words = lines.as(Encoders.STRING())
  30.             .flatMap(new FlatMapFunction<String, String>() {
  31.                 @Override
  32.                 public Iterator<String> call(String line) throws Exception {
  33.                     return Arrays.asList(line.split(" ")).iterator();
  34.                 }
  35.             }, Encoders.STRING());
  36.  
  37.         Dataset<Row> wordCounts = words.groupBy("value").count();
  38.  
  39.         StreamingQuery query = wordCounts.writeStream()
  40.             .outputMode("complete")
  41.             .format("console")
  42.             .start();
  43.         
  44.         query.awaitTermination();
  45.     }
  46. }

 

以上代码编写完成之后，在node3节点执行“nc -lk 9999”启动socket服务器，然后启动代码，向socket中输入以下数据：

  1. 第一次输入：a b c
  2. 第二次输入：d a c
  3. 第三次输入：a b c

可以看到控制台打印如下结果：

  1. -------------------------------------------
  2. Batch: 1
  3. -------------------------------------------
  4. +-----+-----+
  5. |value|count|
  6. +-----+-----+
  7. |    c|    1|
  8. |    b|    1|
  9. |    a|    1|
  10. +-----+-----+
  11.  
  12. -------------------------------------------
  13. Batch: 2
  14. -------------------------------------------
  15. +-----+-----+
  16. |value|count|
  17. +-----+-----+
  18. |    d|    1|
  19. |    c|    2|
  20. |    b|    1|
  21. |    a|    2|
  22. +-----+-----+
  23.  
  24. -------------------------------------------
  25. Batch: 3
  26. -------------------------------------------
  27. +-----+-----+
  28. |value|count|
  29. +-----+-----+
  30. |    d|    1|
  31. |    c|    3|
  32. |    b|    2|
  33. |    a|    3|
  34. +-----+-----+
  35.

三、以上代码注意点如下

  • SparkSQL 默认并行度为200，这里由于数据量少，可以将并行度通过参数“spark.sql.shuffle.partitions”设置少一些。
  • StructuredStreaming读取过来数据默认是DataFrame，默认有“value”名称的列
  • 对获取的DataFrame需要通过as[String]转换成Dataset进行操作
  • 结果输出时的OutputMode有三种输出模式：Complete Mode、Append Mode、Update Mode。
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。