spark 流处理的几个实例

最近写了几个简单的spark structured streaming  的代码实例。 目的是熟悉spark 开发环境搭建， spark 代码开发流程。
开发环境：
系统：win 11 
java : 1.8
scala:2.13 
工具：idea 2022.2  ，maven 3, git 2.37
spark : 3.3.2
一， 使用 spark 结构化流读取文件数据，并做单词统计。
代码：

1. package org.example;
3. import org.apache.spark.sql.Dataset;
4. import org.apache.spark.sql.Row;
5. import org.apache.spark.sql.SparkSession;
6. import org.apache.spark.sql.streaming.OutputMode;
7. import org.apache.spark.sql.streaming.StreamingQuery;
8. import org.apache.spark.sql.streaming.StreamingQueryException;
9. import org.apache.spark.sql.types.DataTypes;
10. import org.apache.spark.sql.types.StructType;
12. import java.util.concurrent.TimeoutException;
14. public class Main {
15.     /*
16.     例子：从文件中读取流， 被定义模式，生成dataset ，使用sql api 进行分析。
17.      */
18.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
19.         System.out.println("Hello world!");
21.         SparkSession spark = SparkSession.builder().appName("spark streaming").config("spark.master", "local")
22.                 .config("spark.sql.warehouse.dir", "file:///app/")
23.                 .getOrCreate();
24.         spark.sparkContext().setLogLevel("ERROR");
26.         StructType schema =
27.                 new StructType().add("empId", DataTypes.StringType).add("empName", DataTypes.StringType)
28.                         .add("department", DataTypes.StringType);
30.         Dataset<Row> rawData = spark.readStream().option("header", false).format("csv").schema(schema)
31.                 .csv("D:/za/spark_data/*.csv");
33.         rawData.createOrReplaceTempView("empData");
35.         Dataset<Row> result = spark.sql("select count(*), department from  empData group by department");
37.         StreamingQuery query = result.writeStream().outputMode("complete").format("console").start();  // 每次触发，全表输出
39.         query.awaitTermination();
42.     }
43. }

复制代码

输出：

二， 使用 spark 结构化流读取socket流，做单词统计，使用Java编程
代码：

1. package org.example;
3. import org.apache.spark.api.java.function.FlatMapFunction;
4. import org.apache.spark.sql.Dataset;
5. import org.apache.spark.sql.Encoders;
6. import org.apache.spark.sql.Row;
7. import org.apache.spark.sql.SparkSession;
8. import org.apache.spark.sql.streaming.StreamingQuery;
9. import org.apache.spark.sql.streaming.StreamingQueryException;
11. import java.util.Arrays;
12. import java.util.concurrent.TimeoutException;
14. public class SocketStreaming_wordcount {
15.     /*
16.      * 从socket 读取字符流，并做word count分析
17.      *
18.      * */
19.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
21.         SparkSession spark = SparkSession
22.                 .builder()
23.                 .appName("JavaStructuredNetworkWordCount")
24.                 .config("spark.master", "local")
25.                 .getOrCreate();
27.         // dataframe 表示 socket 字符流
28.         Dataset<Row> lines = spark
29.                 .readStream()
30.                 .format("socket")
31.                 .option("host", "localhost")
32.                 .option("port", 9999)
33.                 .load();
35. //  把一行字符串切分为 单词
36.         Dataset<String> words = lines
37.                 .as(Encoders.STRING())
38.                 .flatMap((FlatMapFunction<String, String>) x -> Arrays.asList(x.split(" ")).iterator(), Encoders.STRING());
40. //  对单词分组计数
41.         Dataset<Row> wordCounts = words.groupBy("value").count();
42. //  开始查询并打印输出到console
43.         StreamingQuery query = wordCounts.writeStream()
44.                 .outputMode("complete")
45.                 .format("console")
46.                 .start();
48.         query.awaitTermination();
51.     }
54. }

复制代码

输出：

二， 使用 spark 结构化流读取socket流，做单词统计，使用scala 编程
代码：

1. package org.example
3. import org.apache.spark.sql.SparkSession
5. object Main {
6.   def main(args: Array[String]): Unit = {
7.     val spark = SparkSession
8.       .builder
9.       .appName("streaming_socket_scala")
10.       .config("spark.master", "local")
11.       .getOrCreate()
13.     import spark.implicits._
15.     // 创建datafram 象征从网络socket 接收流
16.     val lines = spark.readStream
17.       .format("socket")
18.       .option("host", "localhost")
19.       .option("port", 9999)
20.       .load()
22.     // 切分一行成单词
23.     val words = lines.as[String].flatMap(_.split(" "))
25.     // 进行单词统计
26.     val wordCounts = words.groupBy("value").count()
28.     // 开始查询并输出
29.     val query = wordCounts.writeStream
30.       .outputMode("complete")
31.       .format("console")
32.       .start()
34.     query.awaitTermination()
36.   }
39. }

复制代码

输出：

 
功能比较简单，代码比较简单，可以在网络上找到很多。  但是也是一个完整的spark结构流代码开发流程。权当熟悉下开发流程。
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！