最近写了几个简单的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 结构化流读取文件数据，并做单词统计。
代码：
package org.example;import org.apache.spark.sql.Dataset;import org.apache.spark.sql.Row;import org.apache.spark.sql.SparkSession;import org.apache.spark.sql.streaming.OutputMode;import org.apache.spark.sql.streaming.StreamingQuery;import org.apache.spark.sql.streaming.StreamingQueryException;import org.apache.spark.sql.types.DataTypes;import org.apache.spark.sql.types.StructType;import java.util.concurrent.TimeoutException;public class Main { /* 例子：从文件中读取流， 被定义模式，生成dataset ，使用sql api 进行分析。 */ public static void main(String[] args) throws TimeoutException, StreamingQueryException { System.out.println("Hello world!"); SparkSession spark = SparkSession.builder().appName("spark streaming").config("spark.master", "local") .config("spark.sql.warehouse.dir", "file:///app/") .getOrCreate(); spark.sparkContext().setLogLevel("ERROR"); StructType schema = new StructType().add("empId", DataTypes.StringType).add("empName", DataTypes.StringType) .add("department", DataTypes.StringType); Dataset rawData = spark.readStream().option("header", false).format("csv").schema(schema) .csv("D:/za/spark_data/*.csv"); rawData.createOrReplaceTempView("empData"); Dataset result = spark.sql("select count(*), department from empData group by department"); StreamingQuery query = result.writeStream().outputMode("complete").format("console").start(); // 每次触发，全表输出 query.awaitTermination(); }}输出：
二，使用 spark 结构化流读取socket流，做单词统计，使用Java编程
代码：
package org.example;import org.apache.spark.api.java.function.FlatMapFunction;import org.apache.spark.sql.Dataset;import org.apache.spark.sql.Encoders;import org.apache.spark.sql.Row;import org.apache.spark.sql.SparkSession;import org.apache.spark.sql.streaming.StreamingQuery;import org.apache.spark.sql.streaming.StreamingQueryException;import java.util.Arrays;import java.util.concurrent.TimeoutException;public class SocketStreaming_wordcount { /* * 从socket 读取字符流，并做word count分析 * * */ public static void main(String[] args) throws TimeoutException, StreamingQueryException { SparkSession spark = SparkSession .builder() .appName("JavaStructuredNetworkWordCount") .config("spark.master", "local") .getOrCreate(); // dataframe 表示 socket 字符流 Dataset lines = spark .readStream() .format("socket") .option("host", "localhost") .option("port", 9999) .load();// 把一行字符串切分为 单词 Dataset words = lines .as(Encoders.STRING()) .flatMap((FlatMapFunction) x -> Arrays.asList(x.split(" ")).iterator(), Encoders.STRING());// 对单词分组计数 Dataset wordCounts = words.groupBy("value").count();// 开始查询并打印输出到console StreamingQuery query = wordCounts.writeStream() .outputMode("complete") .format("console") .start(); query.awaitTermination(); }}输出：
二，使用 spark 结构化流读取socket流，做单词统计，使用scala 编程
代码：
package org.exampleimport org.apache.spark.sql.SparkSessionobject Main { def main(args: Array[String]): Unit = { val spark = SparkSession .builder .appName("streaming_socket_scala") .config("spark.master", "local") .getOrCreate() import spark.implicits._ // 创建datafram 象征从网络socket 接收流 val lines = spark.readStream .format("socket") .option("host", "localhost") .option("port", 9999) .load() // 切分一行成单词 val words = lines.as[String].flatMap(_.split(" ")) // 进行单词统计 val wordCounts = words.groupBy("value").count() // 开始查询并输出 val query = wordCounts.writeStream .outputMode("complete") .format("console") .start() query.awaitTermination() }}输出：
功能比较简单，代码比较简单，可以在网络上找到很多。 但是也是一个完整的spark结构流代码开发流程。权当熟悉下开发流程。
—一——步—–一 ——个—–脚——–印———-