ToB企服应用市场:ToB评测及商务社交产业平台

标题: spark 流处理的几个实例 [打印本页]
作者: 北冰洋以北    时间: 2023-4-7 04:00
标题: 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结构流代码开发流程。权当熟悉下开发流程。
 

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



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4