devbox
天景科技苑
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spark实时（五）：InputSource数据源案例演示

作者：天景科技苑 | 2024-07-27 01:23:06

赞

踩

Spark实时（五）：InputSource数据源案例演示

文章目录

InputSource数据源案例演示

一、​​​​​​​File Source

1、读取text文件

2、读取csv文件

3、读取json文件

二、Socket Source 

三、Rate Source

InputSource数据源案例演示

在Spark2.0版本之后，DataFrame和Dataset可以表示静态有边界的数据，也可以表示无边界的流式数据。在Structured Streaming中我们可以使用SparkSession针对流式数据源创建对应的Dataset或者DataFrame，并可以像处理批数据一样使用各种Operators操作处理流式数据。

Structured Streaming的数据源目前支持File Source 、Socket Source 、Rate Source、Kafka Source ，与Kafka的整合在后续整理，这里对其他三种数据源分别演示。

一、​​​​​​​​​​​​​​File Source

Sturctured Streaming可以读取写入目录的文件作为数据流，文件将按照文件修改时间的顺序进行处理，文件必须原子性的存入到监控目录中，支持的格式有text、csv、json、orc、parquet。

1、读取text文件

Scala代码如下：

  1. package com.lanson.structuredStreaming.source
  2.  
  3. import org.apache.spark.sql.streaming.StreamingQuery
  4. import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
  5.  
  6. /**
  7.   *  Structured Streaming监控目录 text格式数据
  8.   */
  9. object SSReadTextData {
  10.   def main(args: Array[String]): Unit = {
  11.  
  12.     //1.创建对象
  13.     val spark: SparkSession = SparkSession.builder().master("local")
  14.       .appName("SSReadTextData")
  15.       .config("spark.sql.shuffle.partitions", 1)
  16.       .getOrCreate()
  17.  
  18.     import  spark.implicits._
  19.  
  20.     spark.sparkContext.setLogLevel("Error")
  21.  
  22.     //2.监控目录
  23.     val ds: Dataset[String] = spark.readStream.textFile("./data/")
  24.  
  25.     val result: DataFrame = ds.map(line => {
  26.       val arr: Array[String] = line.split("-")
  27.       (arr(0).toInt, arr(1), arr(2).toInt)
  28.     }).toDF("id", "name", "age")
  29.  
  30.     val query: StreamingQuery = result.writeStream
  31.       .format("console")
  32.       .start()
  33.  
  34.     query.awaitTermination()
  35.  
  36.   }
  37.  
  38. }

 结果：

Java代码如下：

  1. package com.lanson.structuredStreaming.source;
  2.  
  3. import java.util.concurrent.TimeoutException;
  4. import org.apache.spark.api.java.function.MapFunction;
  5. import org.apache.spark.sql.Dataset;
  6. import org.apache.spark.sql.Encoders;
  7. import org.apache.spark.sql.Row;
  8. import org.apache.spark.sql.SparkSession;
  9. import org.apache.spark.sql.streaming.StreamingQueryException;
  10. import scala.Tuple3;
  11.  
  12. public class SSReadTextData01 {
  13.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
  14.         //1.创建对象
  15.         SparkSession spark = SparkSession.builder().master("local")
  16.                 .appName("SSReadSocketData01")
  17.                 .config("spark.sql.shuffle.partitions", 1)
  18.                 .getOrCreate();
  19.  
  20.         spark.sparkContext().setLogLevel("Error");
  21.  
  22.         Dataset<String> ds = spark.readStream().textFile("./data/");
  23.  
  24.         Dataset<Tuple3<Integer, String, Integer>> ds2 = ds.map(new MapFunction<String, Tuple3<Integer, String, Integer>>() {
  25.             @Override
  26.             public Tuple3<Integer, String, Integer> call(String line) throws Exception {
  27.                 String[] arr = line.split("-");
  28.  
  29.                 return new Tuple3<>(Integer.valueOf(arr[0]), arr[1],Integer.valueOf(arr[2]) );
  30.             }
  31.         }, Encoders.tuple(Encoders.INT(), Encoders.STRING(), Encoders.INT()));
  32.  
  33.         Dataset<Row> result = ds2.toDF("id", "name", "age");
  34.  
  35.         result.writeStream()
  36.                 .format("console")
  37.                 .start()
  38.                 .awaitTermination();
  39.  
  40.     }
  41. }

 结果：

以上代码编写完成之后，向监控的目录“./data”中不断写入含有以下内容的文件，可以看到控制台有对应的流数据输出，这里一定是原子性的将文件复制到对应目录下。文件内容如下：

  1. 1-zhangsan-18
  2. 2-lisi-19
  3. 3-ww-20

2、读取csv文件

Scala代码如下：

  1. package com.lanson.structuredStreaming.source
  2.  
  3. import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
  4. import org.apache.spark.sql.streaming.StreamingQuery
  5. import org.apache.spark.sql.types.StructType
  6.  
  7. /**
  8.   * Structured Streaming 读取CSV数据
  9.   */
  10. object SSReadCsvData {
  11.   def main(args: Array[String]): Unit = {
  12.     //1.创建对象
  13.     val spark: SparkSession = SparkSession.builder().master("local")
  14.       .appName("SSReadCsvData")
  15.       .config("spark.sql.shuffle.partitions", 1)
  16.       .getOrCreate()
  17.  
  18.     import  spark.implicits._
  19.  
  20.     spark.sparkContext.setLogLevel("Error")
  21.  
  22.     //2.创建CSV数据schema
  23.     val userSchema: StructType = new StructType().add("id", "integer")
  24.       .add("name", "string")
  25.       .add("gender", "string")
  26.       .add("age", "integer")
  27.  
  28.  
  29.     val result: DataFrame = spark.readStream
  30.       .option("sep", ",")
  31.       .schema(userSchema)
  32.       .csv("./data/")
  33.  
  34.     val query: StreamingQuery = result.writeStream
  35.       .format("console")
  36.       .start()
  37.  
  38.     query.awaitTermination()
  39.  
  40.   }
  41.  
  42. }

结果：

Java代码如下

  1. package com.lanson.structuredStreaming.source;
  2.  
  3. import java.util.concurrent.TimeoutException;
  4. import org.apache.spark.sql.Dataset;
  5. import org.apache.spark.sql.Row;
  6. import org.apache.spark.sql.SparkSession;
  7. import org.apache.spark.sql.streaming.StreamingQueryException;
  8. import org.apache.spark.sql.types.StructType;
  9.  
  10. /**
  11.  * Structured Streaming 读取CSV数据
  12.  */
  13.  
  14. public class SSReadCsvData01 {
  15.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
  16.         //1.创建对象
  17.         SparkSession spark = SparkSession.builder().master("local")
  18.                 .appName("SSReadCsvData")
  19.                 .config("spark.sql.shuffle.partitions", 1)
  20.                 .getOrCreate();
  21.  
  22.         spark.sparkContext().setLogLevel("Error");
  23.  
  24.         StructType userSchema = new StructType()
  25.                 .add("id", "integer")
  26.                 .add("name", "string")
  27.                 .add("gender", "string")
  28.                 .add("age", "integer");
  29.         Dataset<Row> result = spark.readStream()
  30.                 .option("sep", ",")
  31.                 .schema(userSchema)
  32.                 .csv("./data/");
  33.  
  34.         result.writeStream()
  35.                 .format("console")
  36.                 .start()
  37.                 .awaitTermination();
  38.  
  39.     }
  40. }

 结果：

以上代码运行之后向对应监控的目录下原子性写入含有数据的csv文件，在控制台可以看到实时监控内容。文件内容如下：

  1. 1,zhangsan,一班,100
  2. 2,lisi,二班,200
  3. 3,wangwu,一班,300
  4. 4,maliu,二班,100
  5. 5,tianqi,三班,100
  6. 6,gaoba,三班,50
  7. 7,zs2,四班,50

3、读取json文件

Scala代码如下：

  1. package com.lanson.structuredStreaming.source
  2.  
  3. import org.apache.spark.sql.{DataFrame, SparkSession}
  4. import org.apache.spark.sql.streaming.StreamingQuery
  5. import org.apache.spark.sql.types.StructType
  6.  
  7. /**
  8.   *  Structured Streaming 监控Json格式数据
  9.   */
  10. object SSReadJsonData {
  11.   def main(args: Array[String]): Unit = {
  12.     //1.创建对象
  13.     val spark: SparkSession = SparkSession.builder().master("local")
  14.       .appName("SSReadCsvData")
  15.       .config("spark.sql.shuffle.partitions", 1)
  16.       .getOrCreate()
  17.  
  18.     import  spark.implicits._
  19.  
  20.     spark.sparkContext.setLogLevel("Error")
  21.  
  22.     //2.创建 json 数据schema
  23.     val userSchema: StructType = new StructType().add("id", "integer")
  24.       .add("name", "string")
  25.       .add("age", "integer")
  26.  
  27.  
  28.  
  29.     val result: DataFrame = spark.readStream
  30.       .schema(userSchema)
  31.       .json("./data/")
  32.  
  33.     val query: StreamingQuery = result.writeStream
  34.       .format("console")
  35.       .start()
  36.  
  37.     query.awaitTermination()
  38.  
  39.   }
  40.  
  41. }

结果：

Java代码如下

  1. package com.lanson.structuredStreaming.source;
  2.  
  3.  
  4. import org.apache.spark.sql.Dataset;
  5. import org.apache.spark.sql.Row;
  6. import org.apache.spark.sql.SparkSession;
  7. import org.apache.spark.sql.streaming.StreamingQuery;
  8. import org.apache.spark.sql.streaming.StreamingQueryException;
  9. import org.apache.spark.sql.types.StructType;
  10. import java.util.concurrent.TimeoutException;
  11.  
  12. /**
  13.  * Structured Streaming实时监控目录中json文件作为数据流
  14.  */
  15. public class SSReadJsonData01 {
  16.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
  17.         //1.创建对象
  18.         SparkSession spark = SparkSession.builder().appName("File Source test")
  19.             .master("local")
  20.             .getOrCreate();
  21.  
  22.         //2.设置日志
  23.         spark.sparkContext().setLogLevel("Error");
  24.  
  25.         //3.设置Schema
  26.         StructType userSchema = new StructType().add("id", "integer")
  27.             .add("name", "string")
  28.             .add("age", "integer");
  29.  
  30.         //4.指定监控目录读取数据json数据
  31.         Dataset<Row> ds = spark.readStream()
  32.             .option("sep", ",")
  33.             .schema(userSchema)
  34.             .json("./data/");
  35.  
  36.         //5.打印数据到控制台
  37.         StreamingQuery query =ds.writeStream()
  38.             .format("console")
  39.             .start();
  40.  
  41.         query.awaitTermination();
  42.  
  43.     }
  44. }

结果：

以上代码启动之后，向监控的目录“./data”下原子写入含有以下内容的json文件，在控制台可以看到实时监控内容。json文件内容如下：

  1. {"id":1,"name":"zs","age":18}
  2. {"id":2,"name":"ls","age":19}
  3. {"id":3,"name":"ww","age":20}
  4. {"id":4,"name":"ml","age":21}

注意：实时监控json格式数据时，创建的Schema 中的字段需要与Json中的属性保持一致，否则在映射成表时，Schema中含有但在Json中没有的属性的字段对应的数据会为null。

二、Socket Source 

读取Socket方式需要指定对应的host和port，读取Socket数据源多用于测试场景，这里不再演示。

可以参考案例：

Spark实时（三）：Structured Streaming入门案例-CSDN博客

三、Rate Source

Rate Source是以每秒指定的行数生成数据，每个输出行包含一个timestamp和value，其中timestamp是一个Timestamp含有信息分配的时间类型，value是从0开始的Long类型的数据，Rate Source式多用于测试。

scala代码如下：

  1. package com.lanson.structuredStreaming.source
  2.  
  3. import org.apache.spark.sql.{DataFrame, SparkSession}
  4.  
  5. /**
  6.   * SSRateSource
  7.   */
  8. object SSRateSource {
  9.   def main(args: Array[String]): Unit = {
  10.     //1.创建对象
  11.     val spark: SparkSession = SparkSession.builder().master("local")
  12.       .appName("rate test")
  13. //      .config("spark.sql.shuffle.partitions", 1)
  14.       .getOrCreate()
  15.  
  16.     val result: DataFrame = spark.readStream
  17.       .format("rate")
  18.       // 配置每秒生成多少行数据，默认1行
  19.       .option("rowsPerSecond", "10")
  20.       .option("numPartitions", 5)
  21.       .load()
  22.     result.writeStream
  23.       .format("console")
  24.       .option("numRows","100")
  25.       .option("truncate","false")
  26.       .start()
  27.       .awaitTermination()
  28.  
  29.   }
  30.  
  31. }

结果：

Java代码如下：

  1. package com.lanson.structuredStreaming.source;
  2.  
  3. import java.util.concurrent.TimeoutException;
  4. import org.apache.spark.sql.Dataset;
  5. import org.apache.spark.sql.Row;
  6. import org.apache.spark.sql.SparkSession;
  7. import org.apache.spark.sql.streaming.StreamingQueryException;
  8.  
  9. public class ssratesource01 {
  10.     public static void main(String[] args) throws TimeoutException, StreamingQueryException {
  11.         //1.创建对象
  12.        SparkSession spark = SparkSession.builder().master("local")
  13.                 .appName("rate test")
  14.                 .getOrCreate();
  15.        spark.sparkContext().setLogLevel("Error");
  16.  
  17.         Dataset<Row> result = spark.readStream()
  18.                 .format("rate")
  19.                 // 配置每秒生成多少行数据，默认1行
  20.                 .option("rowsPerSecond", "10")
  21.                 .option("numPartitions", 5)
  22.                 .load();
  23.  
  24.         result.writeStream()
  25.                 .format("console")
  26.                 .option("numRows","100")
  27.                 .option("truncate","false")
  28.                 .start()
  29.                 .awaitTermination();
  30.     }
  31. }

结果： 

 

推荐阅读
相关标签

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