Flink DataStream API编程指南

程序员文章站 2022-07-14 13:42:16

...

Flink中的DataStream程序是实现数据流转换的常规程序(例如：filtering, updating state, defining windows, aggregating)。数据流最初是从各种来源创建的 (例如 message queues, socket streams, files)。结果通过sink返回, 通过sink可以将数据写入文件或者是标准输出(例如：命令行终端)， Flink程序可以运行在各种上下文环境中，standalone 或者嵌入到其他程序中。可以在本地的JVM中运行，也可以在集群中运行。

有关Flink API的基本概念，请参阅 basic concepts。

为了创建你自己的Flink DataStream程序，建议从剖析Flink程序开始，逐步添加您自己的stream transformations。其余部分请参考其他操作和高级特性章节。

实例程序(Example Program)

下面是一个完整的流式窗口word count应用程序，每个5秒会统计来自web socket中的数据。

package com.demon.app

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Auther: fc.w
  * @Date: 2019/1/31
  */
object ExampleProgramApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream("localhost", 9999)

    val counts = text.flatMap(_.toLowerCase.split("\\W+").filter(_.nonEmpty))
      .map((_, 1))
      .keyBy(0)
      .timeWindow(Time.seconds(5))
      .sum(1)

    counts.print()

    env.execute("Window Stream WordCount")
  }

}

要运行示例程序，首先需要从终端启动输入流:

nc -lk 9999

仅需要输入一些单词然后点击回车键，就会输出到应用程序中。如果想看大于1的单词统计，在5秒内反复输入相同的单词（如果你输入不了那么快，可以增加时间窗口的长度）

数据源(Data Sources)

Sources是程序读取数据的入口，可以通过StreamExecutionEnvironment.addSource(sourceFunction) 添加一个source源到程序中。Flink中包含了已实现的源函数，但也可以通过实现SourceFunction 接口自定义非并行源函数，或者通过实现ParallelSourceFunction接口或继承RichParallelSourceFunction来扩展自定义并行源函数。

Flink中有几个预定义的流源，可以StreamExecutionEnvironment访问:

基于文件：

readTextFile(path)： 逐行读取文本文件，遵守TextInputFormat规范的文件，并将其作为字符串返回。
readFile(fileInputFormat, path)： 按照指定文件的输入格式读取（一次）文件。
readFile(fileInputFormat, path, watchType, interval, pathFilter)： 这个方法是前两个的内部调用，根据给定的fileInputFormat读取路径中的文件。根据所提供的watchType（每隔多少ms）定期监控新数据的路径(fileprocessingmode.process_continue)，或者一次性处理当前路径中的数据并退出(FileProcessingMode.PROCESS_ONCE)，使用pathFilter，用户可以进一步排除正在处理的文件。

实现：

在底层，Flink将文件读取过程分为两个子任务，即目录监控和数据读取。这些子任务都是单独实现的。监控是由一个单一的实现(并行度= 1)而非并行任务，数据读取是由多个并行运行的任务执行的。后者的并行度等于作业的并行度。单个监控任务的作用是根据watchType定期或仅扫描一次来监控目录，扫描要处理的文件，将他们分割，并将这些split后的数据分配给下游读取器。每个split仅有一个读取器读取，而 a reader可以逐个地阅读多个拆分。

注意事项：

如果watchType设置为FileProcessingMode.PROCESS_CONTINUOUSLY，当文件被修改时，它的内容将被完全重新处理。这将破坏“exactly-once”语义，因为在文件末尾追加数据将导致对其所有内容进行重新处理。
如果watchType设置为FileProcessingMode.PROCESS_ONCE，Source 文件扫描路径一次并退出，无需等待读取器完成对文件内容的读取。当然读取器将继续读取，直到读取所有文件内容。关闭Source将导致此后不再有checkpoints。这可能会导致节点失败后恢复速度变慢，因为作业将从最后一个检查点恢复读取。

基于scoket:

socketTextStream -从socket读取。元素可以用分隔符分隔。

基于Collection：

fromCollection(Seq)：从Java.util.Collection创建数据流。集合中所有元素必须是相同的类型。
fromCollection(Iterator)： 从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
fromElements(elements: _*)： 从给定的对象序列中创建数据流。所有对象必须是相同的类型。
fromParallelCollection(SplittableIterator) ：并行地从迭代器创建数据流，该类指定迭代器返回的元素的数据类型。
generateSequence(from, to) ：在给定的区间内并行生成数字序列。
addSource()： 附加一个新的源函数，例如，要从Apache Kafka读取数据，可以使用addSource(new FlinkKafkaConsumer08<>(…))。有关详细信息，请参见连接器。

DataStream Transformations

关于可用的流转换，请参考operators

Data Sinks

Data sinks消费DataStreams然后转发给files, sockets, external systems和print。Flink提供了各种内置的输出格式，这些格式封装在DataStreams操作之后:

writeAsText() / TextOutputFormat： 以字符串的形式逐行写入元素。字符串是通过调用每个元素的toString()方法获得的。
writeAsCsv(...) / CsvOutputFormat ：将元组写入以逗号分隔的value文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。
print() / printToErr() ：在标准输出/标准错误流上print每个元素的toString()值。还可以选择在输出之前增加prefix(msg)来帮助区分不同的打印调用。如果并行度大于1，输出还将加上生成输出的任务的标识符。
writeUsingOutputFormat() / FileOutputFormat： 方法和基类自定义文件输出，支持自定义对象到字节的转换。
writeToSocket： 根据SerializationSchema将元素写入Socket。
addSink： 调用自定义sink函数,Flink附带了到其他系统(如Apache Kafka)的连接器，这些连接器实现了sink函数。

注意，DataStream上的write*()方法主要用于调试。他们没有参与Flink的检查点，这意味着这些函数通常具有at-least-once的语义。目标系统的数据刷新依赖于OutputFormat的实现。这意味着并非所有发送到OutputFormat的元素都立即显示在目标系统中。此外，在失败的情况下，这些记录可能会丢失。

为了可靠、精确地将流一次传递到文件系统中，使用flink-connector-filesystem。另外，通过.addsink(…)方法的自定义实现可以参与Flink的检查点，实现精确的一次语义。

Iterations

迭代流程序实现一个步骤函数并将其嵌入IterativeStream中。因为一个DataStream程序可能永远不会结束，因此没有最大迭代次数。相反，您需要指定流的哪些部分反馈给迭代，哪部分使用split 转换或filter被转发到下游。在这里，我们有一个Iteration示例，其中主要部分(重复计算的部分)是一个简单的map()转换，通过使用filter将元素转发到下游。

val iteratedStream = someDataStream.iterate(
  iteration => {
    val iterationBody = iteration.map(/* this is executed many times */)
    (iterationBody.filter(/* one part of the stream */), iterationBody.filter(/* some other part of the stream */))
})

例如下面的程序，它不断地从一系列整数中减去1，直到0:

package com.demon.app

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Auther: fc.w
  * @Date: 2019/1/31
  */
object ExampleProgramApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val someIntegers = env.generateSequence(0, 1000)
    val iteratedStream = someIntegers.iterate(iteration => {
      val minusOne = iteration.map(v => v - 1)
      val stillGreaterThanZero = minusOne.filter(_ > 0)
      val lessThanZero = minusOne.filter(_ <= 0)
      (stillGreaterThanZero, lessThanZero)
    })
    iteratedStream.print()

    env.execute("Window Stream WordCount")
  }

}

执行参数

StreamExecutionEnvironment包含ExecutionConfig，为Job设置指定配置值。
有关更多的参数说明，请参考 execution configuration这些参数主要适用于DataStream API:

setAutoWatermarkInterval(long milliseconds)： 设置Watermark自动触发间隔，也可以通过getAutoWatermarkInterval()获取一个long值。

容错

State & Checkpointing描述如何启用和配置Flink的检查点机制。

Controlling Latency（控制延迟）

默认情况下，元素不会在网络上逐个传输(这会导致不必要的网络流量)，而是被缓冲。缓冲区的大小(实际上是在机器之间传输的)可以在Flink配置文件中设置。虽然这种方法可以很好地优化吞吐量，但当传入流的速度不够快时，它可能会导致延迟问题。为了控制吞吐量和延迟，可以在执行环境(或单个操作符)上使用env.setBufferTimeout(timeoutMillis)设置缓冲区被填满的最大等待时间，在此之后，即使缓冲区没有满，也会自动发送缓冲区。此超时的默认值是100 ms。
用法：

val env: LocalStreamEnvironment = StreamExecutionEnvironment.createLocalEnvironment
env.setBufferTimeout(timeoutMillis)

env.generateSequence(1,10).map(myMap).setBufferTimeout(timeoutMillis)

package com.demon.app

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Auther: fc.w
  * @Date: 2019/1/31
  */
object ExampleProgramApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    /* 分别可以在执行环境和单个操作上使用 */ 
    // 执行环境
    env.setBufferTimeout(100)

    val someIntegers = env.generateSequence(0, 1000)
    val iteratedStream = someIntegers.iterate(iteration => {
      val minusOne = iteration.map(v => v - 1)
      val stillGreaterThanZero = minusOne.filter(_ > 0)
      val lessThanZero = minusOne.filter(_ <= 0)
      (stillGreaterThanZero, lessThanZero)
    })
    // 单个操作
    iteratedStream.setBufferTimeout(100)

    env.execute("Window Stream WordCount")
  }

}

为了最大限度地提高吞吐量，设置setBufferTimeout(-1)，它将移除超时，缓冲区只有在满时才刷新。
为了最小化延迟，将超时设置为接近0的值(例如5或10 ms)。应该避免缓冲区超时为0，因为这会导致严重的性能下降。

故障排除（Debugging）

在分布式集群中运行流程序之前，最好确保所实现的算法能够正常工作。因此，数据分析程序通常是检查结果、调试和改进的增量过程。

Flink提供了简化数据分析程序开发过程的特性。支持IDE中的本地调试、测试数据的注入和结果数据的收集。
本节给出了一些简化Flink程序开发示例。

本地执行环境(Local Execution Environment)

LocalStreamEnvironment在它创建的JVM进程中启动Flink系统。如果从IDE启动LocalEnvironment，可以在代码中设置断点并轻松调试程序。
创建并使用LocalEnvironment如下所示:

val env = StreamExecutionEnvironment.createLocalEnvironment()

val lines = env.addSource(/* some source */)
// build your program

env.execute()

收集数据来源(Collection Data Sources)

Flink提供了由Java集合支持的特殊数据源，以简化测试。一旦对程序进行了测试，sources和sinks就可以轻松地替换或读写外部系统的sources和sinks。

采集数据源的使用方法如下:

val env = StreamExecutionEnvironment.createLocalEnvironment()

// Create a DataStream from a list of elements
val myInts = env.fromElements(1, 2, 3, 4, 5)

// Create a DataStream from any Collection
val data: Seq[(String, Int)] = ...
val myTuples = env.fromCollection(data)

// Create a DataStream from an Iterator
val longIt: Iterator[Long] = ...
val myLongs = env.fromCollection(longIt)

注意:当前，集合数据源要求数据类型和迭代器实现Serializable。此外，收集数据源不能并行执行(并行度= 1)。

迭代器数据接收(Iterator Data Sink)

Flink还提供了一个接收器，用于DataStream结果的收集和测试。它可以使用如下:

import org.apache.flink.streaming.experimental.DataStreamUtils
import scala.collection.JavaConverters.asScalaIteratorConverter

val myResult: DataStream[(String, Int)] = ...
val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.javaStream).asScala

注意:Flink 1.5.0中删除了Flink-streaming-contrib模块。它的类已经被转移到flink-streaming-java和flink-streaming-scala中。

接下来：

操作符(Operators): 可用流操作符的规范。
事件时间(Event Time): Flink的时间概念介绍。
状态和容错(State & Fault Tolerance): 说明如何开发有状态应用程序。
连接器(Connectors): 可用输入和输出连接器的描述。

上一篇： Flink-on-yarn

下一篇： AdaBoost集成方法

Flink DataStream API编程指南

实例程序(Example Program)

数据源(Data Sources)

Flink中有几个预定义的流源，可以StreamExecutionEnvironment访问:

基于文件：

实现：

注意事项：

基于scoket:

基于Collection：

DataStream Transformations

Data Sinks

Iterations

执行参数

容错

Controlling Latency（控制延迟）

故障排除（Debugging）

本地执行环境(Local Execution Environment)

收集数据来源(Collection Data Sources)

迭代器数据接收(Iterator Data Sink)

接下来：

Flink实战(六) - Table API & SQL编程

Flink入门（五）——DataSet Api编程指南

Flink DataStream API之Operators

Flink DataStream API编程指南

DataStream API编程指南之Data Sources（七）

Flink DataStream API - Data Sources获取数据源

Flink DataStream API 介绍

五 Flink DataStream API

Flink DataStream API——Transform

Flink的DataStream API