本文共 3496 字，大约阅读时间需要 11 分钟。

Flink 流处理API 指南

1. 执行环境

1.1 获取执行环境

getExecutionEnvironment 方法用于创建一个执行环境，表示当前执行程序的上下文。如果程序是独立调用的，则返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则返回此集群的执行环境。

val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

如果没有设置并行度，默认会根据 flink-conf.yaml 中的配置，通常默认为 1。

1.2 创建本地执行环境

返回本地执行环境，需要在调用时指定默认的并行度。

val env = StreamExecutionEnvironment.createLocalEnvironment(1)

1.3 创建远程执行环境

返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包。

val env = ExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, "YOURPATH//wordcount.jar")

2. 数据源

2.1 从集合读取数据

定义样例类，传感器 ID、时间戳、温度。

object SensorReading(id: String, timestamp: Long, temperature: Double)

val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream1 = env.fromCollection(List(  SensorReading("sensor_1", 1547718199, 35.8),  SensorReading("sensor_6", 1547718201, 15.4),  SensorReading("sensor_7", 1547718202, 6.7),  SensorReading("sensor_10", 1547718205, 38.1)))stream1.print("stream1:").setParallelism(1)env.execute()

2.2 从文件读取数据

读取指定文件路径的数据。

val stream2 = env.readTextFile("YOUR_FILE_PATH")

2.3 从 Kafka 读取数据

配置 Kafka 的服务器地址和消费者组 ID。

val properties = new Properties()properties.setProperty("bootstrap.servers", "hadoop01:9092")properties.setProperty("group.id", "consumer-group")val stream3 = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

2.4 自定义数据源

通过传入自定义的 SourceFunction 来创建数据源。

val stream4 = env.addSource(new MySensorSource())

class MySensorSource extends SourceFunction[SensorReading] {  var running: Boolean = true  def cancel(): Unit = { running = false }  def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {    val rand = new Random()    var curTemp = 1.to(10).map(i => ("sensor_" + i, 65 + rand.nextGaussian() * 20))    while (running) {      curTemp = curTemp.map(t => (t._1, t._2 + rand.nextGaussian()))      val curTime = System.currentTimeMillis()      curTemp.forEach(t => ctx.collect(SensorReading(t._1, curTime, t._2)))      Thread.sleep(100)    }  }}

3. 数据变换

3.1 map

将数据流中的每个元素应用函数处理。

val streamMap = stream.map(x => x * 2)

3.2 flatMap

将数据流中的每个元素拆分为多个子流处理。

val streamFlatMap = stream.flatMap(x => x.split(" "))

3.3 filter

对数据流进行过滤。

val streamFilter = stream.filter(x => x == 1)

3.4 keyBy

将数据流转换为键控流，按键进行聚合。

DataStream → KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的。

3.5 滚动聚合

这些算子可以针对 KeyedStream 的每一个支流做聚合。

• sum()
• min()
• max()
• minBy()
• maxBy()

3.6 Reduce

将键控流聚合为数据流，合并当前元素和上次聚合结果。

val stream2 = env.readTextFile("YOUR_PATH\\sensor.txt").map(data => {  val dataArray = data.split(",")  SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)}).keyBy("id").reduce((x, y) => SensorReading(x.id, x.timestamp + 1, y.temperature))

3.7 Split 和 Select

将数据流拆分为多个子流或选择特定字段。

val splitStream = stream2.split(sensorData => {  if (sensorData.temperature > 30) Seq("high") else Seq("low")})val high = splitStream.select("high")val low = splitStream.select("low")val all = splitStream.select("high", "low")

3.8 Connect 和 CoMap

将两个数据流连接，保持数据类型不变。

val warning = high.map(sensorData => (sensorData.id, sensorData.temperature))val connected = warning.connect(low)val coMap = connected.map(warningData => (warningData._1, warningData._2, "warning"), lowData => (lowData.id, "healthy"))

3.9 Union

将多个数据流合并为一个数据流。

val unionStream: DataStream[StartUpLog] = appStoreStream.union(otherStream)unionStream.print("union:::")

Connect 与 Union 的区别

• Union：可以处理多个流，需流类型相同。
• Connect：只能处理两个流，在 CoMap 中调整为相同类型。

转载地址：http://hpxp.baihongyu.com/