Structure Streaming-通过 Batch 模式创建 Kafka 工作流

墨蓝 2023-02-20 03:17 4阅读 0赞

这种模式一般需要设置消费的其实偏移量和结束偏移量, 如果不设置 checkpoint 的情况下, 默 认起始偏移量 earliest, 结束偏移量为 latest.  
该模式为一次性作业(批处理), 而非持续性的处理数据.

import org.apache.spark.sql.SparkSession
    
    object Kafkabatch { 
      def main(args: Array[String]): Unit = { 
        val spark = SparkSession
          .builder()
          .master("local[*]")
          .appName("KafkaSource")
          .getOrCreate()
        import spark.implicits._
    
        val df = spark.read
          .format("kafka")
          .option("kafka.bootstrap.servers","hadoop102:9092,hadoop103:9092,hadoop104:9092")
          .option("subscribe","topic1")
          .option("startingOffsets", "earliest")
          .option("endingOffsets", "latest")
          .load
          // .select("value")
            .selectExpr("cast(value as string)")
            .as[String]
            .flatMap(_.split(" "))
            .groupBy("value")
            .count()
    
        df.write
          .format("console")
          .option("truncate",false)
          .save()
      }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
一次性读完

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200622231235678.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70