整合Kafka消费点击流消息

在application.conf中添加配置

# Kafka click_log topic名称
input.topic.click_log="ods_chb_click_log"

读取配置

val `input.topic.click_log` = config.getString("input.topic.click_log")

创建ClickLogETL类，整合Kafka

• 在etl包下创建ClickLogETL，从MQBaseETL继承
• 实现process方法，整合Kafka，并测试打印消费数据

/**
  * 点击流处理逻辑
  */
class ClickLogETL(env:StreamExecutionEnvironment) extends BaseETL[String] {
  /**
    * 业务处理接口
    */
  override def process(): Unit = {
      // 1. 整合kafka
    val clickLogDS: DataStream[String] = getKafkaDataStream(GlobalConfigUtil.`input.topic.click_log`)
    clickLogDS.print()
  }
}

修改App测试

1、在App对象中创建ClickLogETL对象，并调用process方法

2、在Kafka中创建ods层购物车topic

bin/kafka-topics.sh --create --zookeeper node1:2181 --topic ods_chb_click_log --replication-factor 3 --partitions 3 

3、在Kafka中启动控制台生产程序

bin/kafka-console-producer.sh --broker-list node1:9092 --topic ods_chb_click_log

4、启动Flink测试

5、在控制台生产程序中贴入以下消息测试Flink是否能够消费成功

2001:980:91c0:1:8d31:a232:25e5:85d 222.68.172.190 - [05/Sep/2010:11:27:50 +0200] "GET /images/my.jpg HTTP/1.1" 404 23617 "http://www.angularjs.cn/A00n" "Mozilla/5.0 (Macintosh; U; Intel Mac OS X 10_6_4; nl-nl) AppleWebKit/533.17.8 (KHTML, like Gecko) Version/5.0.1 Safari/533.17.8"