Flink 入门Demo详解

一.引言：

Apach Flink 是全新的流处理系统，在Spark Straming的基础上添加了很多特性，主要在于其提供了基于时间和窗口计算的算子，并且支持有状态的存储和 Checkpoint 的重启机制，下面假设有多个温度传感器持续传输当前温度，Flink流处理需要每一段时间提供该时间段内的传感器平均温度。

二.依赖支持

项目是基于maven的scala项目，主要导入flink的scala依赖，如果是java需要另一套依赖：

1.scala

1<properties>
2    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
3    <flink.version>1.7.1</flink.version>
4    <scala.binary.version>2.12</scala.binary.version>
5    <scala.version>2.12.8</scala.version>
6    <hadoop.version>2.6.0</hadoop.version>
7</properties>
8
9<dependencies>
10    <!-- Apache Flink dependencies -->
11    <!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
12    <dependency>
13        <groupId>org.apache.flink</groupId>
14        <artifactId>flink-scala_${scala.binary.version}</artifactId>
15        <version>${flink.version}</version>
16        <scope>provided</scope>
17    </dependency>
18    <dependency>
19        <groupId>org.apache.flink</groupId>
20        <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
21        <version>${flink.version}</version>
22        <scope>provided</scope>
23    </dependency>
24
25    <!-- Scala Library, provided by Flink as well. -->
26    <dependency>
27        <groupId>org.scala-lang</groupId>
28        <artifactId>scala-library</artifactId>
29        <version>${scala.version}</version>
30        <scope>provided</scope>
31    </dependency>
32</dependencies>

2.java

1<dependency>
2    <groupId>org.apache.flink</groupId>
3    <artifactId>flink-streaming-java_2.11</artifactId>
4    <version>${flink.version}</version>
5</dependency>
6<dependency>
7    <groupId>org.apache.flink</groupId>
8    <artifactId>flink-streaming-scala_2.11</artifactId>
9    <version>${flink.version}</version>
10</dependency>

三.辅助类

1.基础温度类 SensorReading

采用case class简化后续处理函数的代码

case class SensorReading(id: String, timestamp: Long, temperature: Double)

2.时间戳提取类

这里采用了Flink的特性: EventTime作为数据的时间戳，通过提取生成SensorReading中的时间戳作为温度传感器传递温度的EventTime

1import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
2import org.apache.flink.streaming.api.windowing.time.Time
3
4/**
5  * Assigns timestamps to SensorReadings based on their internal timestamp and
6  * emits watermarks with five seconds slack.
7  * 根据传感器内部时间戳和传感器读取分配时间戳。
8  */
9class SensorTimeAssigner extends BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(5)) {
10
11  /** Extracts timestamp from SensorReading. */
12  override def extractTimestamp(r: SensorReading): Long = r.timestamp
13
14}

如果是Java，写法稍有不同

1public static class SensorTimeAssigner extends BoundedOutOfOrdernessTimestampExtractor<SensorReading> {
2
3    public SensorTimeAssigner() {
4        super(Time.seconds(5));
5    }
6
7    @Override
8    public long extractTimestamp(SensorReading r) {
9        return r.timestamp;
10    }
11}

3.自定义source类

SparkStreaming采用的是复写Receiver函数实现自定义数据源，通过receiver的store生成数据；Storm通过覆盖Spout的nextTruple方法，emit生成数据；这里Flink通过集成SoureFunction实现run方法，通过collect方法生成数据，这几种流式处理器在自定义数据流这方面其实大致比较类似，换汤不换药。相关的注释都写在代码里了，这里逻辑比较简单，只是通过Random类，去随机模拟温度函数，如果自己有场景需求需要自定义数据源时，可以把Random看做是自己的Socket，在run方法初始化数据源生成数据即可，这里生成数据可以通过Flink Env设置并行度，并行的接收数据，前提是你的数据源支持并行接收。

1import java.util.Calendar
2
3import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction
4import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
5
6import scala.util.Random
7
8/**
9  * Flink 源功能，用于生成具有随机温度值的传感器读取。
10  *
11  * 自定义源方法需要实现run方法与cancel方法即可，需要初始化的连接放到run方法之内即可
12  *
13  * 源的每个并行实例模拟 10 个传感器，这些传感器发出一个传感器
14  * 每 100 ms 阅读一次。
15  *
16  * 注意：这是一个简单的数据生成源函数，不检查其状态。
17  * 如果发生故障，源不会重播任何数据。
18  */
19// 继承时需要定义生成的DStream的类型
20class SensorSource extends RichParallelSourceFunction[SensorReading] {
21
22  // flag indicating whether source is still running.
23  // 指示源是否仍在运行。
24  var running: Boolean = true
25
26  /** run() continuously emits SensorReadings by emitting them through the SourceContext. */
27  override def run(srcCtx: SourceContext[SensorReading]): Unit = {
28    // SourceContext 通过 collect 方法不断向flink发出数据
29    // initialize random number generator
30    val rand = new Random()
31
32    // 获取当前parallel subtask的下标
33    val taskIdx = this.getRuntimeContext.getIndexOfThisSubtask
34
35    // initialize sensor ids and temperatures
36    // 初始化温度转换器 IndexSeq[String，Int] 序列长度为10 华氏度65+
37    var curFTemp = (1 to 10).map {
38      i => ("sensor_" + (taskIdx * 10 + i), 65 + (rand.nextGaussian() * 20))
39    }
40
41    // emit data until being canceled
42    while (running) {
43
44      // update temperature
45      // 更新温度
46      curFTemp = curFTemp.map( t => (t._1, t._2 + (rand.nextGaussian() * 0.5)) )
47      // get current time
48      val curTime = Calendar.getInstance.getTimeInMillis
49
50      // emit new SensorReading
51      // id 区分传感器分区 curTime 标识eventTime temperature 标识温度
52      curFTemp.foreach( t => srcCtx.collect(SensorReading(t._1, curTime, t._2)))
53
54      // wait for 100 ms
55      Thread.sleep(100)
56    }
57
58  }
59
60  /** Cancels this SourceFunction. */
61  override def cancel(): Unit = {
62    running = false
63  }
64
65}

四.主类

主类主要提供三个逻辑：

=> 定义Flink Env 配置相关环境变量，这里定义使用EventTime作为处理时间，其他还有ProcessTime，IngestionTime

1    val env = StreamExecutionEnvironment.getExecutionEnvironment
2    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
3    env.getConfig.setAutoWatermarkInterval(1000L)

=> 获取数据源 Source 并设置 EventTime 的获取方式，Source获取数据源，assign设置事件时间

1    val sensorData: DataStream[SensorReading] = env
2      .addSource(new SensorSource)
3      .assignTimestampsAndWatermarks(new SensorTimeAssigner)

=> 定义数据处理方式并提交任务，这里采用了时间窗口的处理方式，还有基于数据量的窗口以及基于处理函数的算子，这里先介绍最基础的

1    val avgTemp: DataStream[SensorReading] = sensorData
2      .map( r => SensorReading(r.id, r.timestamp, (r.temperature - 32) * (5.0 / 9.0)) )
3      .keyBy(_.id)
4      .timeWindow(Time.seconds(1))
5      .apply(new TemperatureAverage)
6
7    avgTemp.print()
8
9    env.execute("Compute average sensor temperature")

1.完整主类

通过KeyBy可以将原始DataStream转换为KeyedStream，这样同一个key的数据都会发往一个窗口进行处理，这里1s生成一个时间窗口用于监测平均温度

1import com.weibo.ug.push.flink.SensorData.{SensorReading, SensorSource, SensorTimeAssigner}
2import org.apache.flink.streaming.api.TimeCharacteristic
3import org.apache.flink.streaming.api.scala._
4import org.apache.flink.streaming.api.scala.function.WindowFunction
5import org.apache.flink.streaming.api.windowing.time.Time
6import org.apache.flink.streaming.api.windowing.windows.TimeWindow
7import org.apache.flink.util.Collector
8
9/** Object that defines the DataStream program in the main() method */
10object AverageSensorReadings {
11
12  /** main() defines and executes the DataStream program */
13  def main(args: Array[String]) 
14
15    // set up the streaming execution environment
16    val env = StreamExecutionEnvironment.getExecutionEnvironment
17
18    // use event time for the application
19    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
20    // configure watermark interval
21    // 自动生成水位线
22    // 也可以通过 assignTimestampsAndWatermarks 函数内的 getCurrentWatermark 获取水位线生成方式
23    env.getConfig.setAutoWatermarkInterval(1000L)
24
25    // ingest sensor stream
26    // 摄取流数据并绑定 eventTime
27    val sensorData: DataStream[SensorReading] = env
28      // SensorSource generates random temperature readings
29      // 自定义数据源
30      .addSource(new SensorSource)
31      // assign timestamps and watermarks which are required for event time
32      // 分配事件时间所需的时间戳和水印 主要从DStream的数据中获取相关的时间戳 extractTimestamp
33      // 最好在事件生成时为数据类型绑定 eventTime
34      .assignTimestampsAndWatermarks(new SensorTimeAssigner)
35
36    val avgTemp: DataStream[SensorReading] = sensorData
37      // convert Fahrenheit to Celsius using an inlined map function
38      // 通过map函数将华氏度转换为摄氏度 这里也可以调用filter过滤认为不需要的传感器参数
39      .map( r =>
40      SensorReading(r.id, r.timestamp, (r.temperature - 32) * (5.0 / 9.0)) )
41      // organize stream by sensorId
42      // 将同一传感器的温度统一在一起 KeyedStream
43      .keyBy(_.id)
44      // group readings in 1 second windows
45      // 1s生成1个窗口 将一个窗口的数据传输并处理 类似 spark streaming 的 interval
46      .timeWindow(Time.seconds(1))
47      // compute average temperature using a user-defined function
48      .apply(new TemperatureAverage)
49
50    // print result stream to standard out
51    avgTemp.print()
52
53    // execute application
54    env.execute("Compute average sensor temperature")
55  }
56}

2.窗口处理函数

窗口处理函数这里需要覆盖apply方法，有一个注意的点就是继承WindowFunction函数的参数和apply方法的参数是不完全一致的，相关参数的注解都在代码注释里，可以大致浏览；Collector作为一个数据发射器，将处理好的类型进行下一步传递，这里主类的处理逻辑比较简单，只调用了print，也可以通过addSink方法继续向下游发送数据，常见的落盘HDFS，或者写到Kafka，Flink都有相关的实现API。

1/** User-defined WindowFunction to compute the average temperature of SensorReadings */
2// WindowFunction 参数分别代表 In Out Key W ，前两个比较好理解 输入输出类型 第三个为key的类型 第四个用于获取当前window参数
3class TemperatureAverage extends WindowFunction[SensorReading, SensorReading, String, TimeWindow] {
4
5  /** apply() is invoked once for each window */
6  // apply方法的参数不完全和window保持相同顺序 分别为key 当前window 本次窗口输入的迭代类型 与输出的Collector
7  override def apply(
8                      sensorId: String,
9                      window: TimeWindow,
10                      values: Iterable[SensorReading],
11                      out: Collector[SensorReading]): Unit = {
12
13    // compute the average temperature
14    // Int Double 代表返回类型 (c,r)中c代表泛型 r代表values中的元素类型
15    val (cnt, sum) = values.foldLeft((0, 0.0))((c, r) => (c._1 + 1, c._2 + r.temperature))
16    val avgTemp = sum / cnt
17
18    // emit a SensorReading with the average temperature
19    // collector通过collect方法输出每个窗口对应时间戳的平均温度
20    out.collect(SensorReading(sensorId, window.getEnd, avgTemp))
21  }
22}