// 从kafka加载数据 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址 "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则 "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则 "group.id" -> "sparkdemo",//消费者组名称 //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费 //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费 //none:表示如果有offset记录从offset记录开始消费,如果没有就报错 "auto.offset.reset" -> "latest", "auto.commit.interval.ms"->"1000",//自动提交的时间间隔 "enable.auto.commit" -> (true: java.lang.Boolean)//是否自动提交 ) val topics = Array("spark_kafka")//要订阅的主题 //使用工具类从Kafka中消费消息 val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的 ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的 )
object kafka_Demo01 { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local[*]").setAppName("kafka_Demo01") val sc = new SparkContext(conf) val ssc = new StreamingContext(sc, Seconds(5)) ssc.checkpoint("data/ckp") // 从kafka加载数据 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址 "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则 "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则 "group.id" -> "sparkdemo",//消费者组名称 //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费 //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费 //none:表示如果有offset记录从offset记录开始消费,如果没有就报错 "auto.offset.reset" -> "latest", "auto.commit.interval.ms"->"1000",//自动提交的时间间隔 "enable.auto.commit" -> (true: java.lang.Boolean)//是否自动提交 ) val topics = Array("spark_kafka")//要订阅的主题 //使用工具类从Kafka中消费消息 val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的 ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的 ) // 处理消息 val infoDS = kafkaDS.map(record => { val topic = record.topic() val partition = record.partition() val offset = record.offset() val key = record.key() val value = record.value() val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}""" info }) // 输出 infoDS.print() ssc.start() ssc.awaitTermination() ssc.stop(true, true) } }
提交代码
// 处理消息 //注意提交的时机:应该是消费完一小批就该提交一次offset,而在DStream一小批的体现是RDD kafkaDS.foreachRDD(rdd => { rdd.foreach(record => { val topic = record.topic() val partition = record.partition() val offset = record.offset() val key = record.key() val value = record.value() val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}""" info println("消费" + info) }) //获取rdd中offset相关的信息:offsetRanges里面就包含了该批次各个分区的offset信息 val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //提交 kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) println("当前批次的数据已消费并手动提交") })
完整代码
object kafka_Demo02 { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local[*]").setAppName("kafka_Demo01") val sc = new SparkContext(conf) val ssc = new StreamingContext(sc, Seconds(5)) ssc.checkpoint("data/ckp") // 从kafka加载数据 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "hadoop102:9092",//kafka集群地址 "key.deserializer" -> classOf[StringDeserializer],//key的反序列化规则 "value.deserializer" -> classOf[StringDeserializer],//value的反序列化规则 "group.id" -> "sparkdemo",//消费者组名称 //earliest:表示如果有offset记录从offset记录开始消费,如果没有从最早的消息开始消费 //latest:表示如果有offset记录从offset记录开始消费,如果没有从最后/最新的消息开始消费 //none:表示如果有offset记录从offset记录开始消费,如果没有就报错 "auto.offset.reset" -> "latest", // "auto.commit.interval.ms"->"1000",//自动提交的时间间隔 "enable.auto.commit" -> (false: java.lang.Boolean)//是否自动提交 ) val topics = Array("spark_kafka")//要订阅的主题 //使用工具类从Kafka中消费消息 val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, //位置策略,使用源码中推荐的 ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //消费策略,使用源码中推荐的 ) // 处理消息 //注意提交的时机:应该是消费完一小批就该提交一次offset,而在DStream一小批的体现是RDD kafkaDS.foreachRDD(rdd => { rdd.foreach(record => { val topic = record.topic() val partition = record.partition() val offset = record.offset() val key = record.key() val value = record.value() val info: String = s"""topic:${topic}, partition:${partition}, offset:${offset}, key:${key}, value:${value}""" info println("消费" + info) }) //获取rdd中offset相关的信息:offsetRanges里面就包含了该批次各个分区的offset信息 val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //提交 kafkaDS.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) println("当前批次的数据已消费并手动提交") }) // 输出 kafkaDS.print() ssc.start() ssc.awaitTermination() ssc.stop(true, true) } }