Flink

1. Flink简介

1.1 初识Flink

Flink项目的理念是：“Apache Flink是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源的有状态的流处理框架”。

Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。

1.2 Flink的重要特点

1.2.1 事件驱动型(Event-driven)

事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。比较典型的就是以kafka为代表的消息队列几乎都是事件驱动型应用。(Flink的计算也是事件驱动型)

 与之不同的就是SparkStreaming微批次，如图:

 事件驱动型：

1.2.2 流与批的世界观

批处理的特点是有界、大量，非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。

流处理的特点是无界、实时, 无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。

在spark的世界观中，一切都是由批次组成的，离线数据是一个大批次，而实时数据是由一个一个无限的小批次组成的。

而在flink的世界观中，一切都是由流组成的，离线数据是有界限的流，实时数据是一个没有界限的流，这就是所谓的有界流和无界流。

无界数据流：

无界数据流有一个开始但是没有结束，它们不会在生成时终止并提供数据，必须连续处理无界流，也就是说必须在获取后立即处理event。对于无界数据流我们无法等待所有数据都到达，因为输入是无界的，并且在任何时间点都不会完成。处理无界数据通常要求以特定顺序（例如事件发生的顺序）获取event，以便能够推断结果完整性。

有界数据流：

有界数据流有明确定义的开始和结束，可以在执行任何计算之前通过获取所有数据来处理有界流，处理有界流不需要有序获取，因为可以始终对有界数据集进行排序，有界流的处理也称为批处理。

1.2.3 分层API

最底层级的抽象仅仅提供了有状态流，它将通过过程函数（Process Function）被嵌入到DataStream API中。底层过程函数（Process Function）与 DataStream API 相集成，使其可以对某些特定的操作进行底层的抽象，它允许用户可以自由地处理来自一个或多个数据流的事件，并使用一致的容错的状态。除此之外，用户可以注册事件时间并处理时间回调，从而使程序可以处理复杂的计算。

实际上，大多数应用并不需要上述的底层抽象，而是针对核心API（Core APIs）进行编程，比如DataStream API（有界或无界流数据）以及DataSet API（有界数据集）。这些API为数据处理提供了通用的构建模块，比如由用户定义的多种形式的转换（transformations），连接（joins），聚合（aggregations），窗口操作（windows）等等。DataSet API 为有界数据集提供了额外的支持，例如循环与迭代。这些API处理的数据类型以类（classes）的形式由各自的编程语言所表示。

Table API 是以表为中心的声明式编程，其中表可能会动态变化（在表达流数据时）。Table API遵循（扩展的）关系模型：表有二维数据结构（schema）（类似于关系数据库中的表），同时API提供可比较的操作，例如select、project、join、group-by、aggregate等。Table API程序声明式地定义了什么逻辑操作应该执行，而不是准确地确定这些操作代码的看上去如何。

尽管Table API可以通过多种类型的用户自定义函数（UDF）进行扩展，其仍不如核心API更具表达能力，但是使用起来却更加简洁（代码量更少）。除此之外，Table API程序在执行之前会经过内置优化器进行优化。

你可以在表与 DataStream/DataSet 之间无缝切换，以允许程序将 Table API 与 DataStream 以及 DataSet 混合使用。

Flink提供的最高层级的抽象是 SQL 。这一层抽象在语法与表达能力上与 Table API 类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SQL查询可以直接在Table API定义的表上执行。

目前Flink作为批处理还不是主流，不如Spark成熟，所以DataSet使用的并不是很多。Flink Table API和Flink SQL也并不完善，大多都由各大厂商自己定制。所以我们主要学习DataStream API的使用。实际上Flink作为最接近Google DataFlow模型的实现，是流批统一的观点，所以基本上使用DataStream就可以了。

2020年12月8日发布的最新版本1.12.0, 已经完成实现了真正的流批一体. 写好的一套代码, 即可以处理流式数据, 也可以处理离线数据. 这个与前面版本的处理有界流的方式是不一样的, Flink专门对批处理数据做了优化处理.

1.3 Spark or Flink

Spark 和 Flink 一开始都拥有着同一个梦想，他们都希望能够用同一个技术把流处理和批处理统一起来，但他们走了完全不一样的两条路前者是以批处理的技术为根本，并尝试在批处理之上支持流计算；后者则认为流计算技术是最基本的，在流计算的基础之上支持批处理。正因为这种架构上的不同，今后二者在能做的事情上会有一些细微的区别。比如在低延迟场景，Spark 基于微批处理的方式需要同步会有额外开销，因此无法在延迟上做到极致。在大数据处理的低延迟场景，Flink 已经有非常大的优势。

Spark和Flink的主要差别就在于计算模型不同。Spark采用了微批处理模型，而Flink采用了基于操作符的连续流模型。因此，对Apache Spark和Apache Flink的选择实际上变成了计算模型的选择，而这种选择需要在延迟、吞吐量和可靠性等多个方面进行权衡。

如果企业中非要技术选型从Spark和Flink这两个主流框架中选择一个来进行流数据处理，我们推荐使用Flink，主（显而）要（易见）的原因为：

 Flink灵活的窗口

 Exactly Once语义保证

这两个原因可以大大的解放程序员, 加快编程效率, 把本来需要程序员花大力气手动完成的工作交给框架完成

2. Flink快速上手

pom依赖

<flink.version>1.13.2</flink.version>


<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.12</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-compress</artifactId>
    <version>1.21</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.29</version>
  </dependency>

  <!--scala-->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-scala_2.12</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.12</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.12</artifactId>
    <version>1.1.0</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>2.3.5</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>2.3.5</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.1.3</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>3.1.3</version>
  </dependency>

Environment、Source、Transform、Sink、Execution Mode

不同数据源的加载方式（重点kafka数据源）

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig

import java.lang.Thread
import java.util.Properties
import scala.util.Random


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

object SourceTest {
  def main(args: Array[String]): Unit = {
    //1 创建environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1) //并行度个数

    //2 加载数据源
    //val streamElements = env.fromElements(1, 2, 3, 4, 5, "hello", 3.14)

    /*//样例类方式加载
    val streamCaseClass = env.fromCollection(List(
      SensorReading("sensor_1", 1698731530, 26.3),
      SensorReading("sensor_2", 1698731530, 26.7),
      SensorReading("sensor_3", 1698731530, 26.9),
      SensorReading("sensor_3", 1698731531, 28.0)
    ))*/

    /*//指定文件夹加载
    val streamTextFile =
      env.readTextFile("resources\\sensor.txt")*/

    //加载socket端口数据
    val streamSocket =
      env.socketTextStream("kb129", 7777)
        .map(_.split(" "))
        .flatMap(x => x)
        .map((_, 1))
        .keyBy(_._1)
        //.sum(1)
        .reduce((x, y) => (x._1 + " " + y._1, x._2 + y._2))

    /*//kafka拉取数据源
    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"kb129:9092")
    prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest")
    prop.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false")
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"sensorGroup1")

    val streamKafka = env.addSource(
      new FlinkKafkaConsumer[String]("sensor", new SimpleStringSchema(), prop)
    ).flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      .reduce((x: (String, Int), y: (String, Int)) => (x._1, x._2 + y._2))*/


    //调用自定义数据源
    //val streamMySource = env.addSource(new MySensorSource)

    //4 输出，也叫下沉
    streamSocket.print("")

    //5 执行
    env.execute("sourcetest")
  }
}

//模拟自定义数据源
class MySensorSource extends SourceFunction[SensorReading] {
  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
    val random = new Random()
    while (true) {
      ctx.collect(
        SensorReading(
          "随机数：" + random.nextInt(),
          System.currentTimeMillis(),
          random.nextDouble()
        )
      )
      Thread.sleep(1000)
    }
  }

  override def cancel(): Unit = {}
}

3. Flink部署

解压、改名

[root@kb129 ~]# tar -xvf /opt/install/flink-1.13.2-bin-scala_2.12.tgz -C /opt/soft/

[root@kb129 soft]# mv ./flink-1.13.2/ flink132

配置环境变量

[root@kb129 flink1132]# vim /etc/profile

#FLINK_HOME

export FLINK_HOME=/opt/soft/flink1132

export PATH=$FLINK_HOME/bin:$PATH

[root@kb129 flink1132]# source /etc/profile

[root@kb129 flink1132]# flink --version

Version: 1.13.2, Commit ID: 5f007ff

启动/关闭flink

[root@kb129 bin]# start-cluster.sh

[root@kb129 bin]# stop-cluster.sh

网页地址：192.168.142.129:8081

4. Flink运行架构

4.1 运行架构

Flink运行时包含2种进程:1个JobManager和至少1个TaskManager

4.1.1 客户端

严格上说, 客户端不是运行和程序执行的一部分, 而是用于准备和发送dataflow到JobManager. 然后客户端可以断开与JobManager的连接(detached mode), 也可以继续保持与JobManager的连接(attached mode)

客户端作为触发执行的java或者scala代码的一部分运行, 也可以在命令行运行:bin/flink run ...

4.1.2.1 JobManager

控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个不同的JobManager所控制执行。

JobManager会先接收到要执行的应用程序，这个应用程序会包括：作业图（JobGraph）、逻辑数据流图（logical dataflow graph）和打包了所有的类、库和其它资源的JAR包。

JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。

而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。

这个进程包含3个不同的组件

JobManager主要作用：

1. 接受客户端请求。

2. 划分任务。

3. 申请资源。

4.1.2.2 ResourceManager

负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager. 注意这个ResourceManager不是Yarn中的ResourceManager, 是Flink中内置的, 只是赶巧重名了而已.

主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。

当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。

4.1.2.3 Dispatcher

负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的JobMaster 组件. Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中可能并不是必需的，这取决于应用提交运行的方式。

4.1.2.4 JobMaster

JobMaster负责管理单个JobGraph的执行.多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster.

4.1.3 TaskManager

Flink中的工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。

启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。

在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。

总结：

1.工作进程，任务都在TaskManager上运行

2.TaskManager中有资源（Slot槽）

3.需要像JobManager进行交互从而进行资源的注册和使用

4.多个TaskManager可以交换数据

5. Flink流处理核心编程

5.1 Environment

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1) //并行度个数

5.2 Source

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig

import java.lang.Thread
import java.util.Properties
import scala.util.Random


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

object SourceTest {
  def main(args: Array[String]): Unit = {
    //1 创建environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1) //并行度个数

    //2 加载数据源
    val streamElements = env.fromElements(1, 2, 3, 4, 5, "hello", 3.14)

    //样例类方式加载
    val streamCaseClass = env.fromCollection(List(
      SensorReading("sensor_1", 1698731530, 26.3),
      SensorReading("sensor_2", 1698731530, 26.7),
      SensorReading("sensor_3", 1698731530, 26.9),
      SensorReading("sensor_3", 1698731531, 28.0)
    ))

    //指定文件夹加载
    val streamTextFile =
      env.readTextFile("resources\\sensor.txt")

    //加载socket端口数据
    val streamSocket =
      env.socketTextStream("kb129", 7777)
        .map(_.split(" "))
        .flatMap(x => x)
        .map((_, 1))
        .keyBy(_._1)
        //.sum(1)
        .reduce((x, y) => (x._1 + " " + y._1, x._2 + y._2))

    //kafka拉取数据源
    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"kb129:9092")
    prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest")
    prop.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false")
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"sensorGroup1")

    val streamKafka = env.addSource(
      new FlinkKafkaConsumer[String]("sensor", new SimpleStringSchema(), prop)
    ).flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      .reduce((x: (String, Int), y: (String, Int)) => (x._1, x._2 + y._2))


    //调用自定义数据源
    val streamMySource = env.addSource(new MySensorSource)

    //4 输出，也叫下沉
    streamSocket.print("")

    //5 执行
    env.execute("sourcetest")
  }
}

//模拟自定义数据源
class MySensorSource extends SourceFunction[SensorReading] {
  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
    val random = new Random()
    while (true) {
      ctx.collect(
        SensorReading(
          "随机数：" + random.nextInt(),
          System.currentTimeMillis(),
          random.nextDouble()
        )
      )
      Thread.sleep(1000)
    }
  }

  override def cancel(): Unit = {}
}

5.3 Transform

5.3.1 map

作用

将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素

参数

lambda表达式或MapFunction实现类

DataStream → DataStream

5.3.2 flatMap

作用

消费一个元素并产生零个或多个元素

参数

FlatMapFunction实现类

DataStream → DataStream

5.3.3 filter

作用

根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃

参数

FlatMapFunction实现类

DataStream → DataStream

5.3.4 keyBy

作用

把流中的数据分到不同的分区中.具有相同key的元素会分到同一个分区中.一个分区中可以有多重不同的key.

在内部是使用的hash分区来实现的.

分组与分区的区别：

分组：是一个逻辑上的划分，按照key进行区分，经过 keyby，同一个分组的数据肯定会进入同一个分区

分区：下游算子的一个并行实例（等价于一个slot），同一个分区内，可能有多个分组

参数

Key选择器函数: interface KeySelector<IN, KEY>

注意: 什么值不可以作为KeySelector的Key:

 没有覆写hashCode方法的POJO, 而是依赖Object的hashCode. 因为这样分组没有任何的意义: 每个元素都会得到一个独立无二的组. 实际情况是:可以运行, 但是分的组没有意义.

 任何类型的数组

DataStream → KeyedStream

5.3.5 shuffle

作用

把流中的元素随机打乱. 对同一个组数据, 每次只需得到的结果都不同.

参数

无

DataStream → DataStream

~~5.3.6 split和select~~

~~已经过时, 在1.12中已经被移除~~

作用

~~在某些情况下，我们需要将数据流根据某些特征拆分成两个或者多个数据流，给不同数据流增加标记以便于从流中取出.~~

~~split用于给流中的每个元素添加标记. select用于根据标记取出对应的元素, 组成新的流.~~

5.3.7 connect

作用

在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。

Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

参数

另外一个流

DataStream[A], DataStream[B] -> ConnectedStreams[A,B]

5.3.8 union

作用

对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream

connect与 union 区别：

1. union之前两个流的类型必须是一样，connect可以不一样

2. connect只能操作两个流，union可以操作多个。

5.3.9 简单滚动聚合算子

常见的滚动聚合算子

sum,

min,

max

minBy,

maxBy

作用

KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream

参数

如果流中存储的是POJO或者scala的样例类, 参数使用字段名

如果流中存储的是元组, 参数就是位置(基于0...)

KeyedStream -> SingleOutputStreamOperator

5.3.10 reduce

作用

一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

为什么还要把中间值也保存下来? 考虑流式数据的特点: 没有终点, 也就没有最终的概念了. 任何一个中间的聚合结果都是值!

参数、

interface ReduceFunction<T>

KeyedStream -> SingleOutputStreamOperator

5.3.11 process

作用

process算子在Flink算是一个比较底层的算子,很多类型的流上都可以调用,可以从流中获取更多的信息(不仅仅数据本身)

5.3.12 对流重新分区的几个算子

 KeyBy

先按照key分组, 按照key的双重hash来选择后面的分区

 shuffle

对流中的元素随机分区

 reblance

对流中的元素平均分布到每个区.当处理倾斜数据的时候, 进行性能优化

 rescale

同 rebalance一样, 也是平均循环的分布数据。但是要比rebalance更高效, 因为rescale不需要通过网络, 完全走的"管道"。

import nj.zb.kb23.source.SensorReading
import org.apache.flink.api.common.functions.{FilterFunction, FlatMapFunction, MapFunction, ReduceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

import java.util

object TransFormTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("kb129", 7777)

    //map方法1:匿名函数
    /*val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )*/

    //map方法2：MapFunction
    /*val dataStream = stream.map(
      new MapFunction[String, SensorReading] {
        override def map(t: String): SensorReading = {
          val arr = t.split(",")
          SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
        }
      }
    )*/

    //map方法3：引用class
    val dataStream = stream.map(new MyMapFunction)

    //filter方法1
    /*val filterStream = dataStream.filter(
      _.id == "sensor_2"
    )*/

    //filter方法2
    /*val filterStream = dataStream.filter(new FilterFunction[SensorReading] {
      override def filter(t: SensorReading): Boolean = {
        if (t.id == "sensor_2") true else false
      }
    })*/

    //filter方法3
    //val filterStream = dataStream.filter(new MyFilterFunction)

    //flatMap方法1
    /*val flatMapStream = dataStream.flatMap(
      new FlatMapFunction[String, SensorReading] {
        override def flatMap(value: String, out: Collector[SensorReading]): Unit = {
          val arr = value.split(",")
          out.collect(SensorReading(arr(0), arr(1).toLong, arr(2).toDouble))
        }
      }
    )*/

    val keyStream = dataStream.keyBy(_.id)
    //val maxStream = keyStream.max("temperature")
    //val maxStream = keyStream.maxBy("temperature")
    //val maxStream = keyStream.max(2)
    //val sumStream = keyStream.sum("temperature")

    //等于minBy
    /*val reduceMinTemp = keyStream.reduce(
      (x, y) => {
        if (x.temperature < y.temperature) x else y
      }
    )*/

    //等于min 方法1匿名函数
    /*val reduceMinTemp = keyStream.reduce(
      (x, y) => {
        if (x.temperature < y.temperature) x
        else {
          SensorReading(x.id, x.timestamp, y.temperature)
        }
      }
    )*/

    //等于min 方法2
    /*val maxStream = keyStream.reduce(
      new ReduceFunction[SensorReading] {
        override def reduce(x: SensorReading, y: SensorReading): SensorReading = {
          if (x.temperature < y.temperature) x
          else {
            SensorReading(x.id, x.timestamp, y.temperature)
          }
        }
      }
    )*/

    //maxStream.print("maxTemp:")



    //flatMapStream.print()

    env.execute("transformTest")
  }
}

class MyMapFunction extends MapFunction[String, SensorReading]{
  override def map(t: String): SensorReading = {
    val arr = t.split(",")
    SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
  }
}

class MyFilterFunction extends FilterFunction[SensorReading]{
  override def filter(value: SensorReading): Boolean = {
      if (value.id == "sensor_2") true else false
  }
}

5.4 Sink

5.4.1 KafkaSink

import nj.zb.kb23.source.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

/**
 * 将flink数据流输出到Kafka中
 */
object KafkaSink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("kb129", 7777)
    val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble).toString
      }
    )

    dataStream.addSink(
      new FlinkKafkaProducer[String](
        "kb129:9092",
        "sensorout",
        new SimpleStringSchema())
    )

    env.execute("kafkaSink")
  }
}

5.4.2 RedisSink

import nj.zb.kb23.source.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
import redis.clients.jedis.Jedis

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object RedisSinkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    /*val stream = env.socketTextStream("kb129", 7777)
    val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )*/

    //redisSource
    val dataStream = env.addSource(new MyRedisSourceFunction)


    //redis配置
    /*val redisConfig =
      new FlinkJedisPoolConfig.Builder()
        .setHost("kb129")
        .setPort(6379)
        .build()
    dataStream.addSink(new RedisSink[SensorReading](redisConfig, new MyRedisMapper))*/

    //dataStream.addSink(new MyRedisSinkFunction)

    //redisSink
    dataStream.addSink(new MyRedisSinkFunction)

    env.execute("redis")
  }
}

//source
class MyRedisSourceFunction extends RichSourceFunction[SensorReading]{
  private var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis("kb129", 6379)
  }

  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
    val value = jedis.hget("sensor", "sensor_1")

    val arr = value.split("\t")

    ctx.collect(SensorReading("sensor_1", arr(1).toLong, arr(0).toDouble))
  }

  override def cancel(): Unit = {
    jedis.close()
  }
}

class MyMysqlSourceFunction extends RichSourceFunction[SensorReading]{
  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
    //private var resultSet: ResultSet = _
    //private var statement: PreparedStatement = _
    //private var connection: Connection = _

    // 创建数据库连接
    val connection = DriverManager.getConnection("jdbc:mysql://kb129:3306/sql50", "root", "123456")
    // 创建PreparedStatement
    val statement = connection.prepareStatement("SELECT * FROM your_table")
    // 执行查询
    val resultSet = statement.executeQuery()

    // 处理查询结果
    while (resultSet.next()) {
      // 解析结果并发送到下游
      val data = parseResultSet(resultSet)
      ctx.collect(data)
    }
  }

  // 解析ResultSet并返回你的数据类型
  private def parseResultSet(resultSet: ResultSet): SensorReading = {
    SensorReading("", 0l, 0d)
  }

  override def cancel(): Unit = {
    //resultSet.close()
    //statement.close()
    //connection.close()
  }
}



//sink
class MyRedisSinkFunction extends RichSinkFunction[SensorReading] {
  private var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis("kb129", 6379)
  }

  override def invoke(value: SensorReading): Unit = {
    jedis.hset("sensor2", value.id, value.temperature + "\t" + value.timestamp)
  }

  override def close(): Unit = {
    jedis.close()
  }
}


class MyRedisMapper extends RedisMapper[SensorReading] {
  override def getCommandDescription: RedisCommandDescription = {

    new RedisCommandDescription(RedisCommand.HSET, "sensor")
  }

  override def getKeyFromData(data: SensorReading): String = {
    data.id
  }

  override def getValueFromData(data: SensorReading): String = {
    data.timestamp + "\t" + data.temperature
  }
}

5.4.3 ElasticsearchSink

5.4.4 自定义Sink

5.4.45 JDBCSink

import nj.zb.kb23.source.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment


import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 * 将flink数据流输出到MySQL中
 */
object JdbcSink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("kb129", 7777)
    val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    dataStream.addSink(new MyJdbcSink)


    env.execute("jdbcSink")
  }
}

class MyJdbcSink extends RichSinkFunction[SensorReading]{//SinkFunction[SensorReading]{
  var connection: Connection = _
  var insertState :PreparedStatement = _
  var updateState :PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    connection = DriverManager.getConnection(
      "jdbc:mysql://kb129:3306/kb23?useSSL=false",
      "root",
      "123456"
    )
    insertState = connection.prepareStatement(
      "INSERT INTO sensor_temp VALUES(?, ?, ?);"
    )
    updateState = connection.prepareStatement(
      "UPDATE sensor_temp SET timestamp = ?, temp = ? WHERE id = ?;"
    )

    println("insertState:" + insertState)
    println("updateState:" + updateState)
  }

  override def invoke(value: SensorReading, context: SinkFunction.Context): Unit = {
    updateState.setLong(1, value.timestamp)
    updateState.setDouble(2, value.temperature)
    updateState.setString(3, value.id)
    val i = updateState.executeUpdate()
    if (i == 0){
      insertState.setString(1, value.id)
      insertState.setLong(2, value.timestamp)
      insertState.setDouble(3, value.temperature)
      insertState.executeUpdate()
    }
  }

  override def close(): Unit = {
    if (insertState != null){
      insertState.close()
    }
    if (updateState != null) {
      updateState.close()
    }
    if (connection != null) {
      connection.close()
    }
  }
}

7. Flink流处理高阶编程

7.1 Flink的window机制

Flink学习（3）——Window（窗口机制）API Flink学习（3）——Window（窗口机制）API_flink开窗函数_常识与偏见的博客-CSDN博客

7.1.1 窗口概述

在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而Window窗口是一种切割无限数据为有限块进行处理的手段。

在Flink中, 窗口(window)是处理无界流的核心. 窗口把流切割成有限大小的多个"存储桶"(bucket), 我们在这些桶上进行计算.

7.1.2 窗口的分类

窗口分为2类:

1. 基于时间的窗口(时间驱动)

2. 基于元素个数的(数据驱动)

7.1.2.1 基于时间的窗口

时间窗口包含一个开始时间戳(包括)和结束时间戳(不包括), 这两个时间戳一起限制了窗口的尺寸.

在代码中, Flink使用TimeWindow这个类来表示基于时间的窗口. 这个类提供了key查询开始时间戳和结束时间戳的方法, 还提供了针对给定的窗口获取它允许的最大时间差的方法(maxTimestamp())

时间窗口又分4种:

 滚动窗口(Tumbling Windows)

滚动窗口有固定的大小, 窗口与窗口之间不会重叠也没有缝隙.比如,如果指定一个长度为5分钟的滚动窗口, 当前窗口开始计算, 每5分钟启动一个新的窗口.

滚动窗口能将数据流切分成不重叠的窗口，每一个事件只能属于一个窗口。

 滑动窗口(Sliding Windows)

与滚动窗口一样, 滑动窗口也是有固定的长度. 另外一个参数我们叫滑动步长, 用来控制滑动窗口启动的频率.

所以, 如果滑动步长小于窗口长度, 滑动窗口会重叠. 这种情况下, 一个元素可能会被分配到多个窗口中

例如, 滑动窗口长度10分钟, 滑动步长5分钟, 则, 每5分钟会得到一个包含最近10分钟的数据.

 会话窗口(Session Windows)

会话窗口分配器会根据活动的元素进行分组. 会话窗口不会有重叠, 与滚动窗口和滑动窗口相比, 会话窗口也没有固定的开启和关闭时间.

如果会话窗口有一段时间没有收到数据, 会话窗口会自动关闭, 这段没有收到数据的时间就是会话窗口的gap(间隔)

我们可以配置静态的gap, 也可以通过一个gap extractor 函数来定义gap的长度. 当时间超过了这个gap, 当前的会话窗口就会关闭, 后序的元素会被分配到一个新的会话窗口

创建原理:

因为会话窗口没有固定的开启和关闭时间, 所以会话窗口的创建和关闭与滚动, 滑动窗口不同. 在Flink内部, 每到达一个新的元素都会创建一个新的会话窗口, 如果这些窗口彼此相距比较定义的gap小, 则会对他们进行合并. 为了能够合并, 会话窗口算子需要合并触发器和合并窗口函数: ReduceFunction, AggregateFunction, or ProcessWindowFunction

 全局窗口(Global Windows)

全局窗口分配器会分配相同key的所有元素进入同一个 Global window. 这种窗口机制只有指定自定义的触发器时才有用. 否则, 不会做任务计算, 因为这种窗口没有能够处理聚集在一起元素的结束点.

7.1.2.2 基于元素个数的窗口（计数窗口）

 按照指定的数据条数生成一个Window，与时间无关

分2类:

 滚动计数窗口

默认的CountWindow是一个滚动窗口，只需要指定窗口大小即可，当元素数量达到窗口大小时，就会触发窗口的执行。

说明:那个窗口先达到3个元素, 哪个窗口就关闭. 不影响其他的窗口.

 滑动计数窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。下面代码中的sliding_size设置为了2，也就是说，每收到两个相同key的数据就计算一次，每一次计算的window范围最多是3个元素。

7.1.3 Window Function

前面指定了窗口的分配器, 接着我们需要来指定如何计算, 这事由window function来负责. 一旦窗口关闭, window function 去计算处理窗口中的每个元素.

window function 可以是ReduceFunction,AggregateFunction,or ProcessWindowFunction中的任意一种.

ReduceFunction,AggregateFunction更加高效, 原因就是Flink可以对到来的元素进行增量聚合 . ProcessWindowFunction （全窗口函数）可以得到一个包含这个窗口中所有元素的迭代器, 以及这些元素所属窗口的一些元数据信息.

ProcessWindowFunction不能被高效执行的原因是Flink在执行这个函数之前, 需要在内部缓存这个窗口上所有的元素

增量聚合函数是来一条计算一条，而全窗口函数则是等到数据都到了再做计算做一次计算。

 ReduceFunction(增量聚合函数----不会改变数据的类型)

 AggregateFunction(增量聚合函数----可以改变数据的类型)

 ProcessWindowFunction(全窗口函数)

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.{SlidingProcessingTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object WindowEventTimeTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)
    val dataStream = inputStream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    //设置时间语义为 事件时间
    val dataStream2 = dataStream.assignTimestampsAndWatermarks(
      WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(3))
        .withTimestampAssigner(
          new SerializableTimestampAssigner[SensorReading] {
            override def extractTimestamp(element: SensorReading,
                                          recordTimestamp: Long): Long = {
              //指定事件时间的字段
              element.timestamp * 1000
            }
          }
        )
    )
    val lateTag = new OutputTag[SensorReading]("laterData")
    val windowStream = dataStream2.keyBy(_.id)
      //.window(TumblingEventTimeWindows.of(Time.seconds(15)))
      .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(3)))
      //.allowedLateness(Time.minutes(1))  //允许最迟到达的延迟（真正关窗时间延迟1分钟）
      .sideOutputLateData(lateTag)//窗户关闭后，到达的数据处理方式(侧输出)

    /*val resultStream = windowStream.reduce(
      (curReduce, newReduce) => {
        SensorReading(curReduce.id, curReduce.timestamp, curReduce.temperature.min(newReduce.temperature))
      }
    )*/

    val resultStream = windowStream.process(new MyEventProcessWindowFunction)
    resultStream.print("result:")
    resultStream.getSideOutput(lateTag).print("later value:")
   
    env.execute("windowEventTimeTest")
  }
}

class MyEventProcessWindowFunction extends ProcessWindowFunction[SensorReading, (String, Double, Long, Long), String, TimeWindow] {
  override def process(
                        key: String,
                        context: Context,
                        elements: Iterable[SensorReading],
                        out: Collector[(String, Double, Long, Long)]): Unit = {
    val window = context.window

    //getStart 开始时间根据第一条数据时间戳计算 滚动：1698731530000 - (1698731530000-0+15000)%15000
    /*滚动窗口开始时间源码，滑动窗口中有for循环
    long start =
      TimeWindow.getWindowStartWithOffset(
        timestamp, (globalOffset + staggerOffset) % size, size);
    public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
            return timestamp - (timestamp - offset + windowSize) % windowSize};*/
    //println("开始时间：" + window.getStart + "\t结束时间：" + window.getEnd)
    var temp = 100.0
    for (elem <- elements) {
      temp = temp.min(elem.temperature)
    }
   
    out.collect((key, temp, window.getStart, window.getEnd))
  }
}

7.2 Keyed vs Non-Keyed Windows

7.3 Flik中的时间语义与WaterMark

Flink中的水位线【精选】Flink中的水位线_flink 水位线-CSDN博客

7.3.1 Flink中的时间语义

在Flink的流式操作中, 会涉及不同的时间概念

7.3.1.1 处理时间(process time)

处理时间是指的执行操作的各个设备的时间

对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟.比如, 一个长度为1个小时的窗口将会包含设备时钟表示的1个小时内所有的数据. 假设应用程序在 9:15am分启动, 第1个小时窗口将会包含9:15am到10:00am所有的数据, 然后下个窗口是10:00am-11:00am, 等等

处理时间是最简单时间语义, 数据流和设备之间不需要做任何的协调. 他提供了最好的性能和最低的延迟. 但是, 在分布式和异步的环境下, 处理时间没有办法保证确定性, 容易受到数据传递速度的影响: 事件的延迟和乱序

在使用窗口的时候, 如果使用处理时间, 就指定时间分配器为处理时间分配器

7.3.1.2 事件时间(event time)

事件时间是指的这个事件发生的时间.

在event进入Flink之前, 通常被嵌入到了event中, 一般作为这个event的时间戳存在.

在事件时间体系中, 时间的进度依赖于数据本身, 和任何设备的时间无关. 事件时间程序必须制定如何产生Event Time Watermarks(水印) . 在事件时间体系中, 水印是表示时间进度的标志(作用就相当于现实时间的时钟).

在理想情况下，不管事件时间何时到达或者他们的到达的顺序如何, 事件时间处理将产生完全一致且确定的结果. 事件时间处理会在等待无序事件(迟到事件)时产生一定的延迟。由于只能等待有限的时间，因此这限制了确定性事件时间应用程序的可使用性。

假设所有数据都已到达，事件时间操作将按预期方式运行，即使在处理无序或迟到的事件或重新处理历史数据时，也会产生正确且一致的结果。例如，每小时事件时间窗口将包含带有事件时间戳的所有记录，该记录落入该小时，无论它们到达的顺序或处理时间。

在使用窗口的时候, 如果使用事件时间, 就指定时间分配器为事件时间分配器

注意:

在1.12之前默认的时间语义是处理时间, 从1.12开始, Flink内部已经把默认的语义改成了事件时间

7.3.3 Flink中的WaterMark

支持event time的流式处理框架需要一种能够测量event time 进度的方式.比如, 一个窗口算子创建了一个长度为1小时的窗口,那么这个算子需要知道事件时间已经到达了这个窗口的关闭时间, 从而在程序中去关闭这个窗口.

事件时间可以不依赖处理时间来表示时间的进度.例如,在程序中,即使处理时间和事件时间有相同的速度, 事件时间可能会轻微的落后处理时间.另外一方面,使用事件时间可以在几秒内处理已经缓存在Kafka中多周的数据, 这些数据可以照样被正确处理,就像实时发生的一样能够进入正确的窗口.

这种在Flink中去测量事件时间的进度的机制就是watermark(水印). watermark作为数据流的一部分在流动, 并且携带一个时间戳t.

一个Watermark(t)表示在这个流里面事件时间已经到了时间t, 意味着此时, 流中不应该存在这样的数据: 他的时间戳t2<=t (时间比较旧或者等于时间戳)

总结：

1、衡量事件时间的进展。

2、是一个特殊的时间戳，生成之后随着流的流动而向后传递。

3、用来处理数据乱序的问题。

4、触发窗口等得计算、关闭。

5、单调递增的（时间不能倒退）。

6、 Flink认为，小于Watermark时间戳的数据处理完了，不应该再出现。

 有序流中的水印

事件是有序的(生成数据的时间和被处理的时间顺序是一致的), watermark是流中一个简单的周期性的标记。

有序场景：

1、底层调用的也是乱序的Watermark生成器，只是乱序程度传了一个0ms。

2、 Watermark = maxTimestamp – outOfOrdernessMills – 1ms

= maxTimestamp – 0ms – 1ms

=>事件时间 – 1ms

乱序流中的水印

在下图中, 按照他们时间戳来看, 这些事件是乱序的, 则watermark对于这些乱序的流来说至关重要.

通常情况下, 水印是一种标记, 是流中的一个点, 所有在这个时间戳(水印中的时间戳)前的数据应该已经全部到达. 一旦水印到达了算子, 则这个算子会提高他内部的时钟的值为这个水印的值.

乱序场景：

1、什么是乱序 => 时间戳大的比时间戳小的先来

2、乱序程度设置多少比较合适？

a) 经验值 => 对自身集群和数据的了解，大概估算。

b) 对数据进行抽样。

c) 肯定不会设置为几小时，一般设为秒或者分钟。

3、 Watermark = maxTimestamp – outOfOrdernessMills – 1ms

=>当前最大的事件时间 – 乱序程度（等待时间）- 1ms

7.4 窗口允许迟到的数据

7.5 侧输出流(sideOutput)

7.5.1 处理窗口关闭之后的迟到数据

7.5.2 使用侧输出流把一个流拆成多个流

split算子可以把一个流分成两个流, 从1.12开始已经被移除了. 官方建议我们用侧输出流来替换split算子的功能.

需求: 采集监控传感器水位值，将水位值高于5cm的值输出到side output

import nj.zb.kb23.source.SensorReading
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("kb129", 7777)
    val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    //分流
    val normalStream = dataStream.process(new SplitTempProcess)
    normalStream.print("normal")

    val lowStream = normalStream.getSideOutput(new OutputTag[(String, Long, Double)]("low"))
    lowStream.print("low")

    val alarmStream = normalStream.getSideOutput(new OutputTag[(String, Long, Double)]("alarm"))
    alarmStream.print("alarm")

    val highStream = normalStream.getSideOutput(new OutputTag[(String, Long, Double)]("high"))
    highStream.print("high")

    env.execute("sideOutputTest")
  }
}

/**
 * 自定义ProcessFunction
 * 分流操作
 */
class SplitTempProcess extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(
                               value: SensorReading,
                               ctx: ProcessFunction[SensorReading, SensorReading]#Context,
                               out: Collector[SensorReading]): Unit = {
    if (value.temperature > 40 && value.temperature < 80) { //合理的温度范围
      out.collect(value)
    } else if (value.temperature <= 40) {
      ctx.output(
        new OutputTag[(String, Long, Double)]("low"),
        (value.id, value.timestamp, value.temperature)
      )
    } else if (value.temperature >= 100) {
      ctx.output(
        new OutputTag[(String, Long, Double)]("alarm"),
        (value.id, value.timestamp, value.temperature)
      )
    } else {
      ctx.output(
        new OutputTag[(String, Long, Double)]("high"),
        (value.id, value.timestamp, value.temperature)
      )
    }
  }
}

7.6 ProcessFunction API(底层API)

Flink之处理函数（ProcessFunction）Flink之处理函数（ProcessFunction）_flink processfunction-CSDN博客

7.6.1 ProcessFunction

最基本的处理函数，基于DataStream直接调用process（）

public abstract class ProcessFunction<I, O> extends AbstractRichFunction {}

7.6.2 KeyedProcessFunction

基于KeyedStream调用process（），使用定时器必须基于KeyedStream

public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction {}

7.6.3 CoProcessFunction

基于ConnectedStream调用process（）

public abstract class CoProcessFunction<IN1, IN2, OUT> extends AbstractRichFunction {}

7.6.4 ProcessJoinFunction

public abstract class ProcessJoinFunction<IN1, IN2, OUT> extends AbstractRichFunction {}

7.6.5 BroadcastProcessFunction

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {}

7.6.6 KeyedBroadcastProcessFunction

public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT>
        extends BaseBroadcastProcessFunction {}

7.6.7 ProcessWindowFunction

基于WindowedStream调用process（）

abstract class ProcessWindowFunction[IN, OUT, KEY, W <: Window]
    extends AbstractRichFunction {}

7.6.8 ProcessAllWindowFunction

基于WindowedAllStream调用process（）

abstract class ProcessAllWindowFunction[IN, OUT, W <: Window]
    extends AbstractRichFunction {}

7.7 定时器

在Flink程序中，为了实现数据的聚合统计，或者开窗计算之类的功能，我们一般都要先用keyBy()算子对数据流进行“按键分区”，得到一个KeyedStream。也就是指疋一个键(key )，按照它的哈希值（hash code）将数据分成不同的“组”，然后分配到不同的并行子任务上执行计算;这相当于做了一个逻辑分流的操作，从而可以充分利用并行计算的优势实时处理海重效据。

另外，只有在KeyedStream中才支持使用TimerService 设置定时器的操作。所以一般情况下，都是先做了keyBy()分区之后，再去定义处理操作;代码中更川常见的处理函数是KeyedProcessFunction，最基本的ProcessFunction反而出镜率没那么高。

基于处理时间或者事件时间处理过一个元素之后, 注册一个定时器, 然后指定的时间执行，定时器只能用于keyedStream中，即keyby之后使用.

Context和OnTimerContext所持有的TimerService对象拥有以下方法:

 currentProcessingTime(): Long 返回当前处理时间

 currentWatermark(): Long 返回当前watermark的时间戳

 registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前key的processing time的定时器。当processing time到达定时时间时，触发timer。

 registerEventTimeTimer(timestamp: Long): Unit 会注册当前key的event time 定时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。

 deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。

 deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。

7.7.1 基于处理时间的定时器

7.7.2 基于事件时间的定时器

import nj.zb.kb23.source.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

/**
 * 监控温度 在指定时间内连续上升/下降 ，如果发生 则产生一条数据流
 */
object TimeServiceTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("localhost", 7777)
    val dataStream = stream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )

    val keyedStream = dataStream.keyBy(_.id)

    //监控跳变温度
    //keyedStream.process(new TempJumpWaring)

    //监控连续上升/下降温度
    val resultStream = keyedStream.process(new TempIncWaring(10))
    //监控到的异常数据，指定输出
    //resultStream.addSink(new MysqlSink())
    resultStream.print("warning")

    env.execute("StateTest")
  }
}

/**
 * @param num 单位：秒
 */
class TempIncWaring(num: Int) extends KeyedProcessFunction[String, SensorReading, String] {
  lazy val lastTmpState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last_tmp", classOf[Double]))
  lazy val firstTagState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("firstTag", classOf[Boolean]))
  lazy val lastTimeState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("lastTime", classOf[Long]))

  override def processElement(
           value: SensorReading,
           ctx: KeyedProcessFunction[String, SensorReading, String]#Context,
           out: Collector[String]): Unit = {
    val bool = firstTagState.value()

    if (!bool){ //判断是否为第一条数据
      firstTagState.update(true)
    } else {
      val lastTmp = lastTmpState.value()
      val timeTS = lastTimeState.value()
      if (value.temperature > lastTmp && timeTS == 0){  //判断当前温度是否高于上次温度 且 判断定时器是否重置
        //注册num秒钟之后的定时器
        val l = ctx.timerService().currentProcessingTime() + num * 1000
        lastTimeState.update(l)
        ctx.timerService().registerProcessingTimeTimer(l)
      } else if (value.temperature <= lastTmp && timeTS != 0){  //如果温度没有上升 且 存在定时器,则删除定时器
        ctx.timerService().deleteProcessingTimeTimer(timeTS)
        //将定时器的时间重置
        lastTimeState.clear()
      }
    }
    lastTmpState.update(value.temperature)

  }

  override def onTimer(
          timestamp: Long,
          ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext,
          out: Collector[String]): Unit = {
    out.collect("定时器stu：温度报警" + ctx.getCurrentKey+ "\t" + timestamp)
    //将定时器的时间重置
    lastTimeState.clear()
  }
}

7.8 Flink状态编程

【精选】Flink-状态编程_flink状态编程-CSDN博客

7.8.1 什么是状态

在流式计算中有些操作一次处理一个独立的事件(比如解析一个事件), 有些操作却需要记住多个事件的信息(比如窗口操作)。

那些需要记住多个事件信息的操作就是有状态的。

流式计算分为无状态计算和有状态计算两种情况。

 无状态的计算观察每个独立事件，并根据最后一个事件输出结果。例如，流处理应用程序从传感器接收水位数据，并在水位超过指定高度时发出警告。

 有状态的计算则会基于多个事件输出结果。以下是一些例子。例如，计算过去一小时的平均水位，就是有状态的计算。所有用于复杂事件处理的状态机。例如，若在一分钟内收到两个相差20cm以上的水位差读数，则发出警告，这是有状态的计算。流与流之间的所有关联操作，以及流与静态表或动态表之间的关联操作，都是有状态的计算。

7.8.2 为什么需要管理状态

下面的几个场景都需要使用流处理的状态功能:

 去重

数据流中的数据有重复，我们想对重复数据去重，需要记录哪些数据已经流入过应用，当新数据流入时，根据已流入过的数据来判断去重。

 检测

检查输入流是否符合某个特定的模式，需要将之前流入的元素以状态的形式缓存下来。比如，判断一个水位传感器数据流中的温度是否在持续上升。

 聚合

对一个时间窗口内的数据进行聚合分析，分析一个小时内水位的情况。

 更新机器学习模型

在线机器学习场景下，需要根据新流入数据不断更新机器学习的模型参数。

7.8.3 Flink中的状态分类

Flink包括两种基本类型的状态Managed State和Raw State

	Managed State	Raw State（了解）
状态管理方式	Flink Runtime托管, 自动存储, 自动恢复, 自动伸缩	用户自己管理
状态数据结构	Flink提供多种常用数据结构, 例如:ListState, MapState等	字节数组: byte[]
使用场景	绝大数Flink算子	所有算子

注意:

1. 从具体使用场景来说，绝大多数的算子都可以通过继承Rich函数类或其他提供好的接口类，在里面使用Managed State。Raw State一般是在已有算子和Managed State不够用时，用户自定义算子时使用。

2. 在我们平时的使用中Managed State已经足够我们使用, 下面重点学习Managed State

7.8.4 Managed State的分类

对Managed State继续细分，它又有两种类型

a) Keyed State(键控状态)。

b) Operator State(算子状态)。

Operator State Keyed State

	Operator State	Keyed State
适用用算子类型	可用于所有算子: 常用于source, 例如 FlinkKafkaConsumer	只适用于KeyedStream上的算子
状态分配	一个算子的子任务对应一个状态	一个Key对应一个State: 一个算子会处理多个Key, 则访问相应的多个State
创建和访问方式	实现CheckpointedFunction或ListCheckpointed(已经过时)接口	重写RichFunction, 通过里面的RuntimeContext访问
横向扩展	并行度改变时有多种重新分配方式可选: 均匀分配和合并后每个得到全量	并发改变, State随着Key在实例间迁移
支持的数据结构	ListState和BroadCastState	ValueState, ListState,MapState ReduceState, AggregatingState

1、值状态（ValueState）

这里需要传入状态的名称和类型——这跟我们声明一个变量时做的事情完全一样。有了这个描述器，运行时环境就可以获取到状态的控制句柄（handler）了。

2、列表状态（ListState）

将需要保存的数据，以列表（List）的形式组织起来。在 ListState接口中同样有一个类型参数 T，表示列表中数据的类型。ListState 也提供了一系列的方法来操作状态，使用方式与一般的 List 非常相似

⚫ Iterable get()：获取当前的列表状态，返回的是一个可迭代类型 Iterable；

⚫ update(List values)：传入一个列表 values，直接对状态进行覆盖；

⚫ add(T value)：在状态列表中添加一个元素 value；

⚫ addAll(List values)：向列表中添加多个元素，以列表 values 形式传入。

类似地，ListState 的状态描述器就叫作 ListStateDescriptor，用法跟 ValueStateDescriptor 完全一致。

3、映射状态（MapState）

把一些键值对（key-value）作为状态整体保存起来，可以认为就是一组 key-value 映射的列表。对应的 MapState接口中，就会有 UK、UV 两个泛型，分别表示保存的 key 和 value 的类型。同样，MapState 提供了操作映射状态的方法，与 Map 的使用非常类似。

⚫ UV get(UK key)：传入一个 key 作为参数，查询对应的 value 值；

⚫ put(UK key, UV value)：传入一个键值对，更新 key 对应的 value 值；

⚫ putAll(Map map)：将传入的映射 map 中所有的键值对，全部添加到映射状态中；

⚫ remove(UK key)：将指定 key 对应的键值对删除；

⚫ boolean contains(UK key)：判断是否存在指定的 key，返回一个 boolean 值。另外，MapState 也提供了获取整个映射相关信息的方法：

⚫ Iterable> entries()：获取映射状态中所有的键值对；

⚫ Iterable keys()：获取映射状态中所有的键（key），返回一个可迭代 Iterable 类型；

⚫ Iterable values()：获取映射状态中所有的值（value），返回一个可迭代 Iterable 类型；

⚫ boolean isEmpty()：判断映射是否为空，返回一个 boolean 值。

4、归约状态（ReducingState）

类似于值状态（Value），不过需要对添加进来的所有数据进行归约，将归约聚合之后的值作为状态保存下来。ReducingState这个接口调用的方法类似于 ListState，只不过它保存的只是一个聚合值，所以调用.add()方法时，不是在状态列表里添加元素，而是直接把新数据和之前的状态进行归约，并用得到的结果更新状态。

归约逻辑的定义，是在归约状态描述器（ReducingStateDescriptor）中，通过传入一个归约函数（ReduceFunction）来实现的。这里的归约函数，就是我们之前介绍 reduce 聚合算子时讲到的 ReduceFunction，所以状态类型跟输入的数据类型是一样的。

这里的描述器有三个参数，其中第二个参数就是定义了归约聚合逻辑的 ReduceFunction，另外两个参数则是状态的名称和类型。

5、聚合状态（AggregatingState）

与归约状态非常类似，聚合状态也是一个值，用来保存添加进来的所有数据的聚合结果。与 ReducingState 不同的是，它的聚合逻辑是由在描述器中传入一个更加一般化的聚合函数（AggregateFunction）来定义的；这也就是之前我们讲过的 AggregateFunction，里面通过一个累加器（Accumulator）来表示状态，所以聚合的状态类型可以跟添加进来的数据类型完全不同，使用更加灵活。

同样地，AggregatingState 接口调用方法也与 ReducingState 相同，调用.add()方法添加元素时，会直接使用指定的 AggregateFunction 进行聚合并更新状态。

7.9 Flink的容错机制

Flink 使用之Checkpoint配置Flink 使用之Checkpoint配置 - 简书

7.9.1 状态的一致性

当在分布式系统中引入状态做checkpoint时，自然也引入了一致性问题。

一致性实际上是"正确性级别"的另一种说法，也就是说在成功处理故障并恢复之后得到的结果，与没有发生任何故障时得到的结果相比，前者到底有多正确？举例来说，假设要对最近一小时登录的用户计数。在系统经历故障之后，计数结果是多少？如果有偏差，是有漏掉的计数还是重复计数？

一致性级别

在流处理中，一致性可以分为3个级别:

 at-most-once(最多变一次):

这其实是没有正确性保障的委婉说法——故障发生之后，计数结果可能丢失。

 at-least-once(至少一次):

这表示计数结果可能大于正确值，但绝不会小于正确值。也就是说，计数程序在发生故障后可能多算，但是绝不会少算。

 exactly-once(严格变一次):

这指的是系统保证在发生故障后得到的计数结果与正确值一致.既不多算也不少算。

曾经，at-least-once非常流行。第一代流处理器(如Storm和Samza)刚问世时只保证at-least-once，原因有二:

1. 保证exactly-once的系统实现起来更复杂。这在基础架构层(决定什么代表正确，以及exactly-once的范围是什么)和实现层都很有挑战性

2. 流处理系统的早期用户愿意接受框架的局限性，并在应用层想办法弥补(例如使应用程序具有幂等性，或者用批量计算层再做一遍计算)。

最先保证exactly-once的系统(Storm Trident和Spark Streaming)在性能和表现力这两个方面付出了很大的代价。为了保证exactly-once，这些系统无法单独地对每条记录运用应用逻辑，而是同时处理多条(一批)记录，保证对每一批的处理要么全部成功，要么全部失败。这就导致在得到结果前，必须等待一批记录处理结束。因此，用户经常不得不使用两个流处理框架(一个用来保证exactly-once，另一个用来对每个元素做低延迟处理)，结果使基础设施更加复杂。曾经，用户不得不在保证exactly-once与获得低延迟和效率之间权衡利弊。Flink避免了这种权衡。

Flink的一个重大价值在于，它既保证了exactly-once，又具有低延迟和高吞吐的处理能力。

从根本上说，Flink通过使自身满足所有需求来避免权衡，它是业界的一次意义重大的技术飞跃。尽管这在外行看来很神奇，但是一旦了解，就会恍然大悟。

端到端的状态一致性

目前我们看到的一致性保证都是由流处理器实现的，也就是说都是在 Flink 流处理器内部保证的；而在真实应用中，流处理应用除了流处理器以外还包含了数据源（例如 Kafka）和输出到持久化系统。

端到端的一致性保证，意味着结果的正确性贯穿了整个流处理应用的始终；每一个组件都保证了它自己的一致性，整个端到端的一致性级别取决于所有组件中一致性最弱的组件。

具体划分如下:

 source端

需要外部源可重设数据的读取位置.

目前我们使用的Kafka Source具有这种特性: 读取数据的时候可以指定offset

 flink内部

依赖checkpoint机制

 sink端

需要保证从故障恢复时，数据不会重复写入外部系统. 有2种实现形式:

a) 幂等（Idempotent）写入

所谓幂等操作，是说一个操作，可以重复执行很多次，但只导致一次结果更改，也就是说，后面再重复执行就不起作用了。

b) 事务性（Transactional）写入

需要构建事务来写入外部系统，构建的事务对应着 checkpoint，等到 checkpoint 真正完成的时候，才把所有对应的结果写入 sink 系统中。对于事务性写入，具体又有两种实现方式：预写日志（WAL）和两阶段提交（2PC）

7.9.2 Checkpoint原理

Flink 使用之Checkpoint配置：Flink 使用之Checkpoint配置 - 简书

Flink具体如何保证exactly-once呢? 它使用一种被称为"检查点"（checkpoint）的特性，在出现故障时将系统重置回正确状态。下面通过简单的类比来解释检查点的作用。

Flink的检查点算法

checkpoint机制是Flink可靠性的基石，可以保证Flink集群在某个算子因为某些原因(如异常退出)出现故障时，能够将整个应用流图的状态恢复到故障之前的某一状态，保证应用流图状态的一致性.

快照的实现算法:

1. 简单算法--暂停应用, 然后开始做检查点, 再重新恢复应用

2. Flink的改进Checkpoint算法. Flink的checkpoint机制原理来自"Chandy-Lamport algorithm"算法(分布式快照算法)的一种变体: 异步 barrier 快照（asynchronous barrier snapshotting）

每个需要checkpoint的应用在启动时，Flink的JobManager为其创建一个 CheckpointCoordinator，CheckpointCoordinator全权负责本应用的快照制作。

理解Barrier

流的barrier是Flink的Checkpoint中的一个核心概念. 多个barrier被插入到数据流中, 然后作为数据流的一部分随着数据流动(有点类似于Watermark).这些barrier不会跨越流中的数据.

每个barrier会把数据流分成两部分: 一部分数据进入当前的快照 , 另一部分数据进入下一个快照 . 每个barrier携带着快照的id. barrier 不会暂停数据的流动, 所以非常轻量级. 在流中, 同一时间可以有来源于多个不同快照的多个barrier, 这个意味着可以并发的出现不同的快照.

7.9.3 Savepoint原理

1. Flink 还提供了可以自定义的镜像保存功能，就是保存点（savepoints）

2. 原则上，创建保存点使用的算法与检查点完全相同，因此保存点可以认为就是具有一些额外元数据的检查点

3. Flink不会自动创建保存点，因此用户（或外部调度程序）必须明确地触发创建操作

4. 保存点是一个强大的功能。除了故障恢复外，保存点可以用于：有计划的手动备份，更新应用程序，版本迁移，暂停和重启应用，等等

7.9.4 checkpoint和savepoint的区别

Savepoint	Checkpoint
Savepoint是由命令触发, 由用户创建和删除	Checkpoint被保存在用户指定的外部路径中
保存点存储在标准格式存储中，并且可以升级作业版本并可以更改其配置。	当作业失败或被取消时，将保留外部存储的检查点。
用户必须提供用于还原作业状态的保存点的路径。	用户必须提供用于还原作业状态的检查点的路径。

7.9.5 flink中的重启策略

Flink是一个分布式流处理框架，它的重启策略是指在任务执行过程中出现异常时，如何进行任务的重启。Flink提供了多种重启策略，可以根据不同的场景选择合适的重启策略。

1. NoRestartStrategy

NoRestartStrategy是Flink默认的重启策略，它表示不进行重启，当任务出现异常时，直接停止任务。这种策略适用于一些不需要保证数据完整性的任务，例如一些数据采集任务。

2. FixedDelayRestartStrategy

FixedDelayRestartStrategy是一种固定延迟的重启策略，当任务出现异常时，会在一定的时间间隔后进行重启。这种策略适用于一些可以容忍一定数据丢失的任务，例如一些日志处理任务。

key	type	default	含义
restart-strategy.fixed-delay.attempts	Integer	1	重试次数如果超过则任务退出
restart-strategy.fixed-delay.delay	Duration	1s	两个任务之间的延迟

3. FailureRateRestartStrategy

FailureRateRestartStrategy是一种根据失败率进行重启的策略，当任务出现异常时，会根据一定的失败率进行重启。这种策略适用于一些需要保证数据完整性的任务，例如一些金融交易任务。

key	type	default	含义
restart-strategy.failure-rate.delay	Duration	1 s	重试的时间间隔
restart-strategy.failure-rate.failure-rate-interval	Duration	1 min	时间间隔
restart-strategy.failure-rate.max-failures-per-interval	Integer	1	时间间隔内最大的错误

4. FallbackRestartStrategy

FallbackRestartStrategy是一种备用重启策略，当其他重启策略都无法满足需求时，会使用FallbackRestartStrategy进行重启。这种策略适用于一些特殊场景，例如一些需要手动干预的任务。

Flink提供了多种重启策略，可以根据不同的场景选择合适的重启策略。在实际应用中，需要根据任务的特点和需求选择合适的重启策略，以保证任务的稳定性和数据的完整性。

7.9.6 flink反压

1. 反压的理解

简单来说，Flink 拓扑中每个节点（Task）间的数据都以阻塞队列的方式传输，下游来不及消费导致队列被占满后，上游的生产也会被阻塞，最终导致数据源的摄入被阻塞。

反压（BackPressure）通常产生于这样的场景：短时间的负载高峰导致系统接收数据的速率远高于它处理数据的速率。许多日常问题都会导致反压，例如，垃圾回收停顿可能会导致流入的数据快速堆积，或遇到大促、秒杀活动导致流量陡增。

2. 反压的危害

反压如果不能得到正确的处理，可能会影响到 checkpoint 时长和 state 大小，甚至可能会导致资源耗尽甚至系统崩溃。

1）影响 checkpoint 时长：barrier 不会越过普通数据，数据处理被阻塞也会导致checkpoint barrier 流经整个数据管道的时长变长，导致 checkpoint 总体时间（End to End Duration）变长。

2）影响 state 大小：barrier 对齐时，接受到较快的输入管道的 barrier 后，它后面数据会被缓存起来但不处理，直到较慢的输入管道的 barrier 也到达，这些被缓存的数据会被放到 state 里面，导致 checkpoint 变大。

这两个影响对于生产环境的作业来说是十分危险的，因为 checkpoint 是保证数据一致性的关键，checkpoint 时间变长有可能导致 checkpoint 超时失败，而 state 大小同样可能拖慢 checkpoint 甚至导致 OOM （使用 Heap-based StateBackend）或者物理内存使用超出容器资源（使用 RocksDBStateBackend）的稳定性问题。因此，我们在生产中要尽量避免出现反压的情况

7.10 广播状态

广播状态让所有并行子任务都持有同一份状态，也就意味着一旦状态有变化，所有子任务上的实例都要更新。

一个最为普遍的应用，就是“动态配置”或者“动态规则”。我们在处理流数据时，有时会基于一些配置(configuration）或者规则（rule)。简单的配置当然可以直接读取配置文件，一次加载，永久有效;但数据流是连续不断的，如果这配置随着时间推移还会动态变化，那又该怎么办呢?一个简单的想法是，定期扫描配置文件，发现改变就立即更新。但这样就需要另外启动一个扫描进程，如果扫描周期太长，配置更新不及时就会导致结果错误;如果扫描周期太短，又会耗费大量资源做无用功。解决的办法，还是流处理的“事件驱动”思路——我们可以将这动态的配置数据看作一条流，将这条流和本身要处理的数据流进行连接(connect)，就可以实时地更新配置进行计算了。

由于配置或者规则数据是全局有效的，我们需要把它广播给所有的并行子任务。而子任务需要把它作为一个算子状态保存起来，以保证故障恢复后处理结果是一致的。这时的状态，就需要把它作为一个算子状态保存起来，以保证故障恢复后处理结果是一致的。这时的状态，就是一个典型的广播状态。我们知道，广播状态与其他算子状态的列表（list）结构不同，底层是以键值对(key-value）形式描述的，所以其实就是一个映射状态(MapState)。

在代码上，可以直接调用DataStream 的 broadcast()方法，传入一个“映射状态描述器”(MapStateDescriptor）说明状态的名称和类型，就可以得到一个“广播流”(BroadcastStream);进而将要处理的数据流与这条广播流进行连接( connect )，就会得到“广播连接流”(BroadcastConnectedStream)。注意广播状态只能用在广播连接流中。

Flink支持广播。可以将数据广播到TaskManager上就可以供TaskManager中的SubTask/task去使用，数据存储到内存中。这样可以减少大量的shuffle操作，而不需要多次传递给集群节点；

比如在数据join阶段，不可避免的就是大量的shuffle操作，我们可以把其中一个dataSet广播出去，一直加载到taskManager的内存中，可以直接在内存中拿数据，避免了大量的shuffle，导致集群性能下降；

可以理解广播就是一个公共的共享变量；将一个数据集广播后，不同的Task都可以在节点上获取到；每个节点只存一份；如果不使用广播，每一个Task都会拷贝一份数据集，造成内存资源浪费；如下图所示：

注意：广播变量是要把dataset广播到内存中，所以广播的数据量不能太大，否则会出现OOM；广播变量的值不可修改，这样才能确保每个节点获取到的值都是一致的；

import org.apache.flink.api.common.state.{MapStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

/**
 * “zs,login”,"ls,login","zs,pay","ls,car","ww,login"
 * 需求:找到 "login,pay" 用户
 */

case class Action(userId: String, action: String)

case class Pattern(action1: String, action2: String)

object BroadcastTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val actionStream =
      env.socketTextStream("localhost", 7777)
        .map(x => Action(x.split(",")(0), x.split(",")(1)))

    val patternStream =
      env.socketTextStream("localhost", 7778)
        .map(x => Pattern(x.split(",")(0), x.split(",")(1)))

    val patternsDescriptor =
      new MapStateDescriptor[Unit, Pattern]("patterns", classOf[Unit], classOf[Pattern])

    val broadcastStream = patternStream.broadcast(patternsDescriptor)

    val connectedStream =
      actionStream
        .keyBy(_.userId)
        .connect(broadcastStream)
        .process(new MyPatternFunction)

    connectedStream.print("match:")

    env.execute("broadcastTest")
  }
}

class MyPatternFunction extends KeyedBroadcastProcessFunction[String, Action, Pattern, (String, String, Pattern)] {
  lazy val patternsDescriptor =
    new MapStateDescriptor[Unit, Pattern]("patterns", classOf[Unit], classOf[Pattern])

  //记录每一位用户的上一个操作行为
  lazy private val preActionState: ValueState[String] =
    getRuntimeContext.getState(new ValueStateDescriptor[String]("prev_action", classOf[String]))

  override def processElement(
                               value: Action,
                               ctx: KeyedBroadcastProcessFunction[String, Action, Pattern, (String, String, Pattern)]#ReadOnlyContext,
                               out: Collector[(String, String, Pattern)]): Unit = {
    println("processElement:" + value)
    val readOnlyBroadcastState = ctx.getBroadcastState(patternsDescriptor)
    val pattern = readOnlyBroadcastState.get(Unit) //pattern为"login, pay"
    if (pattern != null) {
      println("匹配规则:" + pattern.action1, pattern.action2)
      val preAction = preActionState.value()
      if (pattern != null &&
        preAction != null &&
        pattern.action1 == preAction &&
        pattern.action2 == value.action
      ) {
        out.collect((ctx.getCurrentKey, value.action, pattern))
      }
    }
    preActionState.update(value.action)
  }

  override def processBroadcastElement(
                                        value: Pattern,
                                        ctx: KeyedBroadcastProcessFunction[String, Action, Pattern, (String, String, Pattern)]#Context,
                                        out: Collector[(String, String, Pattern)]): Unit = {
    println("processBroadcastElement:" + value)
    val bcState = ctx.getBroadcastState(patternsDescriptor)
    bcState.put(Unit, value)
  }
}