Flink实时数据分析系统开发

前边我们已经开发完毕了上报服务系统, 我们可以通过上报服务系统把电商页面中的点击流数据发送到Kafka中, 那么接下来我们就来开发Flink实时分析系统, 通过流的方式读取Kafka中的消息, 进而分析数据。

业务

• 实时分析频道热点
• 实时分析频道PV/UV
• 实时分析频道新鲜度
• 实时分析频道地域分布
• 实时分析运营商平台
• 实时分析浏览器类型
  技术
• Flink实时处理算子
• 使用CheckPoint和水印解决Flink生产上遇到的问题（网络延迟、丢数据）
• Flink整合Kafka
• Flink整合HBase

搭建【Flink实时数据分析系统】项目环境

导入Maven项目依赖

1. 将pom.xml文件中的依赖导入到real-process项目的pom.xml

  <properties>
        <scala.version>2.11</scala.version>
        <flink.version>1.6.1</flink.version>
        <hadoop.version>2.7.5</hadoop.version>
        <hbase.version>2.0.0</hbase.version>
    </properties>


    <dependencies>
        <!--kafka 客户端-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_${
    scala.version}</artifactId>
            <version>0.10.1.0</version>
        </dependency>


        <!--flink对接kafka：导入flink使用kafka的依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.10_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>

        <!--批处理-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>
        <!--导入scala的依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>

        <!--模块二 流处理-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>

        <!--数据落地flink和hbase的集成依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-hbase_${
    scala.version}</artifactId>
            <version>${
    flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>${
    hbase.version}</version>
        </dependency>


        <!--<dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>2.0.0</version>
        </dependency>-->


        <!--hbase依赖于hadoop-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${
    hadoop.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>${
    hadoop.version}</version>
            <!--xml.parser冲突 flink hdfs-->
            <exclusions>
                <exclusion>
                    <groupId>xml-apis</groupId>
                    <artifactId>xml-apis</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${
    hadoop.version}</version>
            <!--数据同步：canal 和 hadoop protobuf-->
            <exclusions>
                <exclusion>
                    <groupId>com.google.protobuf</groupId>
                    <artifactId>protobuf-java</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <!--对象和json 互相转换的-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.44</version>
        </dependency>
    </dependencies>


    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.itheima.realprocess.App</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

3. real-process模块添加scala支持
4. main和test创建scala文件夹，并标记为源代码和测试代码目录

创建项目包结构

包名	说明
com.itheima.realprocess.util	存放工具类
com.itheima.realprocess.bean	存放实体类
com.itheima.realprocess.task	存放具体的分析任务 每一个业务都是一个任务，对应的分析处理都写在这里

导入实时系统Kafka/Hbase配置

1. application.conf导入到resources`目录

#
#kafka的配置
#
# Kafka集群地址
bootstrap.servers="hadoop102:9092,hadoop103:9092,hadoop104:9092"
# ZooKeeper集群地址
zookeeper.connect="hadoop102:2181,hadoop103:2181,hadoop104:2181"
# Kafka Topic名称
input.topic="pyg"
# 消费组ID
group.id="pyg"
# 自动提交拉取到消费端的消息offset到kafka
enable.auto.commit="true"
# 自动提交offset到zookeeper的时间间隔单位（毫秒）
auto.commit.interval.ms="5000"
# 每次消费最新的数据
auto.offset.reset="latest"

2. log4j.properties到resources目录下

log4j.rootLogger=warn,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

注意修改`kafka服务器`和`hbase服务器`的机器名称

获取配置文件API介绍

ConfigFactory.load()介绍

• 使用ConfigFactory.load()可以自动加载配置文件中的application.conf文件（注意：名字一定不要写错，否则无法加载），并返回一个Config对象
• 使用Config对象可以获取到配置文件中的配置项
• application.conf文件是一个properties文件，存放key-value键值对的数据。
  常用API

方法名	说明
getString(“key”)	获取配置文件中指定key的值对应的字符串
getInt(“key”)	获取配置文件中指定key的值对应的整型数字
getLong(“key”)	同上
getBoolean(“key”)	同上
##3 编写scala代码读取配置工具类
在com.itheima.realprocess.util包下创建GlobalConfigUtil单例对象（object）
步骤

1. 使用ConfigFactory.load获取配置对象
2. 编写方法加载application.conf配置
3. 添加一个main方法测试，工具类是否能够正确读取出配置项。
   示例代码：
   GlobalConfigUtil.scala

/**
  * 加载配置文件工具类
  */
object GlobalConfigUtil {
    

  private val config: Config = ConfigFactory.load()

  /*------------------
   * Kafka配置
   *------------------*/

  val bootstrapServers = config.getString("bootstrap.servers")
  val zookeeperConnect = config.getString("zookeeper.connect")
  val inputTopic = config.getString("input.topic")
  val groupId = config.getString("group.id")
  val enableAutoCommit = config.getString("enable.auto.commit")
  val autoCommitIntervalMs = config.getString("auto.commit.interval.ms")
  val autoOffsetReset = config.getString("auto.offset.reset")
    
  
  // 测试配置文件读取类
  def main(args: Array[String]): Unit = {
    
    println(bootstrapServers)
    println(zookeeperConnect)
    println(inputTopic)
    println(groupId)
    println(enableAutoCommit)
    println(autoCommitIntervalMs)
    println(autoOffsetReset)
  }
}

初始化Flink流式计算环境

步骤

1. 创建App单例对象
2. 创建main方法，获取StreamExecutionEnvironment运行环境
3. 设置流处理的时间为EventTime，使用数据发生的时间来进行数据处理
4. 设置Flink的并行度
5. 编写测试代码，测试Flink程序是否能够正确执行
   参考代码

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._

object App {
    
  def main(args: Array[String]): Unit = {
    

    //初始化Flink的流式环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //设置处理的时间为EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //设置并行度
    env.setParallelism(1)

    //本地测试  加载本地集合 成为一个DataStream 打印输出
    val localDataStream = env.fromCollection(
      List("hadoop", "hive", "hbase", "flink")
    )

    localDataStream.print()

    //执行任务
    env.execute("real-process")
  }

}

注意：
1. 一定要导入`import org.apache.flink.api.scala._`隐式转换，否则Flink程序无法执行
2. 到导入`org.apache.flink.streaming.api`下的`TimeCharacteristic`，否则`没有EventTime`

Flink添加checkpoint容错支持

Checkpoint是Flink实现容错机制最核心的功能，它能够根据配置周期性地基于Stream中各个Operator的状态来生成Snapshot，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些Snapshot进行恢复，从而修正因为故障带来的程序数据状态中断。
步骤

1. Flink envrionment中添加checkpoint支持
2. 运行Flink程序测试checkpoint是否配置成功（检查HDFS中是否已经保存snapshot数据）
   实现
   在Flink流式处理环境中，添加以下checkpoint的支持，确保Flink的高容错性，数据不丢失。

//
// 保证程序长时间运行的安全性进行checkpoint操作
//
// 5秒启动一次checkpoint
env.enableCheckpointing(5000)
// 设置checkpoint只checkpoint一次
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// 设置两次checkpoint的最小时间间隔
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000)
// checkpoint超时的时长
env.getCheckpointConfig.setCheckpointTimeout(60000)
// 允许的最大checkpoint并行度
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
// 当程序关闭的时，触发额外的checkpoint
env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
// 设置checkpoint的地址
env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-checkpoint/"))

3. 启动HDFS
4. 启动Flink程序测试
5. 如果测试成功，在HDFS中应该生成如下几个目录

Flink整合Kafka

Flink读取Kafa数据

步骤

1. 配置Kafka连接属性
2. 使用FlinkKafkaConsumer010整合Kafka
3. 添加一个source到当前Flink环境
4. 启动zookeeper
5. 启动kafka
6. 运行Flink程序测试是否能够从Kafka中消费到数据
   实现

1.配置Kafka连接属性

//
// 整合Kafka
//
val properties = new Properties()

// # Kafka集群地址
properties.setProperty("bootstrap.servers",GlobalConfigUtil.bootstrapServers)
// # ZooKeeper集群地址
properties.setProperty("zookeeper.connect",GlobalConfigUtil.zookeeperConnect)
// # Kafka Topic名称
properties.setProperty("input.topic",GlobalConfigUtil.inputTopic)
// # 消费组ID
properties.setProperty("group.id",GlobalConfigUtil.groupId)
// # 自动提交拉取到消费端的消息offset到kafka
properties.setProperty("enable.auto.commit",GlobalConfigUtil.enableAutoCommit)
// # 自动提交offset到zookeeper的时间间隔单位（毫秒）
properties.setProperty("auto.commit.interval.ms",GlobalConfigUtil.autoCommitIntervalMs)
// # 每次消费最新的数据
properties.setProperty("auto.offset.reset",GlobalConfigUtil.autoOffsetReset)

val consumer = new FlinkKafkaConsumer010[String](
    GlobalConfigUtil.inputTopic,
    new SimpleStringSchema(),
    properties
)

2.添加一个source到当前Flink环境

val kafkaDataStream: DataStream[String] = env.addSource(consumer)

3.打印DataStream中的数据

kafkaDataStream.print()

1. 启动zookeeper

2. 启动kafka

3. 运行Flink程序

4. 运行上报服务系统

5. 启动消息生成器, 测试是否能够从Kafka中消费到数据
   如果Flink从Kafka消费成功会打印以下数据，就证明我们的代码是正确的。

  {
    
      "count": 1,
      "message": "{\"browserType\":\"谷歌浏览器\",\"categoryID\":6,\"channelID\":4,\"city\":\"America\",\"country\":\"china\",\"entryTime\":1544601660000,\"leaveTime\":1544634060000,\"network\":\"联通\",\"produceID\":4,\"province\":\"china\",\"source\":\"百度跳转\",\"userID\":13}",
      "timeStamp": 1553188417573
  }

Kafka消息解析为元组

步骤

• 使用map算子，遍历kafka中消费到的数据
• 使用FastJSON转换为JSON对象
• 将JSON的数据解析成一个元组
• 打印map映射后的元组数据
• 测试是否能够正确解析
  代码

1. 使用map算子，将kafka中消费到的数据，使用FastJSON准换为JSON对象
2. 将JSON的数据解析成一个元组

object App {
    
  def main(args: Array[String]): Unit = {
    
    ...
      
    //JSON->元组
    val tupleDataStream = kafkaDataStream.map {
    
      msgJSON => {
    
        val jsonObject = JSON.parseObject(msgJSON)

        val message = jsonObject.getString("message")
        val count = jsonObject.getLong("count")
        val timeStamp = jsonObject.getLong("timeStamp")

        (message, count, timeStamp)
      }
    }
    tupleDataStream.print()
    
    //执行任务
    env.execute("real-process")
  }

}

3. 打印经过map映射后的元组数据，测试是否能够正确解析
   原来的格式
   
   解析后的格式
   

Flink封装点击流消息为样例类

步骤

1. 创建一个ClickLog样例类中来封装消息
2. 使用map算子将数据封装到ClickLog样例类
   代码
3. 在bean包中，创建ClickLog样例类，添加以下字段

• 频道ID(channelID)
• 产品类别ID(categoryID)
• 产品ID(produceID)
• 国家(country)
• 省份(province)
• 城市(city)
• 网络方式(network)
• 来源方式(source)
• 浏览器类型(browserType)
• 进入网站时间(entryTime)
• 离开网站时间(leaveTime)
• 用户的ID(userID)

1. 在ClickLog伴生对象中实现apply方法
2. 使用FastJSON的JSON.parseObject方法将JSON字符串构建一个ClickLog实例对象
3. 使用map算子将数据封装到ClickLog样例类
4. 在样例类中编写一个main方法，传入一些JSON字符串测试是否能够正确解析
5. 重新运行Flink程序，测试数据是否能够完成封装

使用JSONObject.get("key")获取到的数据是一个Any类型，要调用toString方法转换为String类型

参考代码
在bean目录下创建

import com.alibaba.fastjson.JSON

/**
 * 使用ClickLog样例类来封装点击流日志
 */
// - 频道ID(channelID)
// - 产品类别ID(categoryID)
// - 产品ID(produceID)
// - 国家(country)
// - 省份(province)
// - 城市(city)
// - 网络方式(network)
// - 来源方式(source)
// - 浏览器类型(browserType)
// - 进入网站时间(entryTime)
// - 离开网站时间(leaveTime)
// - 用户的ID(userID)
case class ClickLog (
                      var channelID:String,
                      var categoryID:String,
                      var produceID:String,
                      var country:String,
                      var province:String,
                      var city:String,
                      var network:String,
                      var source:String,
                      var browserType:String,
                      var entryTime:String,
                      var leaveTime:String,
                      var userID:String)
object ClickLog {
    
  def apply(json:String): ClickLog = {
    
    val jsonObject = JSON.parseObject(json)

    val channelID = jsonObject.getString("channelID")
    val categoryID = jsonObject.getString("categoryID")
    val produceID = jsonObject.getString("produceID")
    val country = jsonObject.getString("country")
    val province = jsonObject.getString("province")
    val city = jsonObject.getString("city")
    val network = jsonObject.getString("network")
    val source = jsonObject.getString("source")
    val browserType = jsonObject.getString("browserType")
    val entryTime = jsonObject.getString("entryTime")
    val leaveTime = jsonObject.getString("leaveTime")
    val userID = jsonObject.getString("userID")

    ClickLog(
      channelID,
      categoryID,
      produceID,
      country,
      province,
      city,
      network,
      source,
      browserType,
      entryTime,
      leaveTime,
      userID
    )
  }

  def main(args: Array[String]): Unit = {
    
      val clickLog = ClickLog("{\"browserType\":\"火狐\",\"categoryID\":20,\"channelID\":19,\"city\":\"ZhengZhou\",\"country\":\"china\",\"entryTime\":1544605260000,\"leaveTime\":1544634060000,\"network\":\"电信\",\"produceID\":7,\"province\":\"HeBeijing\",\"source\":\"百度跳转\",\"userID\":6}")
      println(clickLog)
    }
}

修改一下 (ClickLog(message), count, timeStamp)

 //JSON->元组
    val tupleDataStream = kafkaDataStream.map {
    
      msgJSON => {
    
        val jsonObject = JSON.parseObject(msgJSON)

        val message = jsonObject.getString("message")
        val count = jsonObject.getLong("count")
        val timeStamp = jsonObject.getLong("timeStamp")

        (ClickLog(message), count, timeStamp)
      }
    }
    tupleDataStream.print()

封装Kafka消息为Message样例类

步骤

1. 创建一个Message样例类，将ClickLog、时间戳、数量封装
2. 将Kafka中的数据整个封装到Message类中
3. 运行Flink测试
   参考代码
   Message.scala

/**
 * 封装Kafka中的消息
 * @param clickLog 点击流浏览数据Bean对象
 * @param count 数量
 * @param timestamp 时间戳
 */
case class Message(var clickLog:ClickLog,
                   var count:Long,
                   var timestamp:Long)

App.scala

object App {
    
  def main(args: Array[String]): Unit = {
    
	...
    // 使用map算子，将kafka中消费到的数据
    val tupleDataStream = kafkaDataStream.map {
    
      msgJSON => {
    
        val jsonObject = JSON.parseObject(msgJSON)

        val message = jsonObject.getString("message")
        val count = jsonObject.getLong("count")
        val timeStamp = jsonObject.getLong("timeStamp")

     //   (ClickLog(message), count, timeStamp)
        Message(ClickLog(message),count,timeStamp)
    }
    }
    ...
  }
}

Flink添加水印支持

水印（watermark）就是一个时间戳，Flink可以给数据流添加水印，可以理解为：Flink收到一条消息后，额外给这个消息添加了一个时间字段，这就是添加水印。

• 水印并不会影响原有Eventtime
• 一般会设置水印时间，比Eventtime小几秒钟
• 当数据流添加水印后，会按照水印时间来触发窗口计算
• 当接收到的水印时间 >= 窗口的endTime，则触发计算
  观察下图:
  60号消息的EventTime为10:10:00, 正好符合10:00:00到10:10:00这个时间窗口的endtime。正常情况, 该消息应该被这个窗口计算的。但是当发生网络延迟的时候，该消息可能会晚到几秒钟，那么当它到达flink时，该窗口已经运算完毕。为了解决该问题，我们为该消息设置watermark时间为10:09:57，当它到达flink时，会把该watermark时间设置为窗口的当前时间，由于小于endtime，所以该消息到达后并不会立即计算。直到一个携带watermark时间大于或者等于endtime的时候，窗口计算才会被触发。这样就可以有效的解决由于网络延迟造成的数据计算不精确的情况。
  
  1.在App.scala中添加水印支持

tupleDataStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Message]{
    
    // 当前时间戳
    var currentTimestamp = 0l
    //延迟时间
    var maxDelayTime = 2000l

    // 获取水印时间
    override def getCurrentWatermark: Watermark = {
    
        new Watermark(currentTimestamp - maxDelayTime)
    }

    // 获取EventTime
    override def extractTimestamp(element: Message, previousElementTimestamp: Long): Long = {
    

        // 比较当前消息的时间戳和上一个消息的时间戳,取最大值
        currentTimestamp = Math.max(element.timeStamp,previousElementTimestamp)
        currentTimestamp
    }
})

2.启动执行测试，观察输出

Message(1,1557475915909,ClickLog(火狐,9,14,ShiJiaZhuang,china,1544612460000,1544634060000,移动,2,HeNan,必应跳转,11))