Flink DataStream API编程指南

Flink中的DataStream程序是实现数据流转换的常规程序（例如，Filter，更新状态，定义窗口，聚合）。最初从各种源（例如，消息队列，套接字流，文件）创建数据流。结果通过接收器返回，接收器可以例如将数据写入文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在许多计算机的集群上执行。

有关FlinkAPI 基本概念的介绍，请参阅基本概念。

为了创建您自己的FlinkDataStream程序，我们鼓励您从Flink程序的解剖开始，逐步添加您自己的流转换。其余部分充当其他 算子操作和高级函数的参考。

示例程序

以下程序是流窗口字数统计应用程序的完整工作示例，它在5秒窗口中对来自Web套接字的单词进行计数。您可以复制并粘贴代码以在本地运行它。

  1. import org.apache.flink.api.common.functions.FlatMapFunction;
  2. import org.apache.flink.api.java.tuple.Tuple2;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.api.windowing.time.Time;
  6. import org.apache.flink.util.Collector;
  8. public class WindowWordCount {
  10.     public static void main(String[] args) throws Exception {
  12.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  14.         DataStream<Tuple2<String, Integer>> dataStream = env
  15.                 .socketTextStream("localhost", 9999)
  16.                 .flatMap(new Splitter())
  17.                 .keyBy(0)
  18.                 .timeWindow(Time.seconds(5))
  19.                 .sum(1);
  21.         dataStream.print();
  23.         env.execute("Window WordCount");
  24.     }
  26.     public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
  27.         @Override
  28.         public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
  29.             for (String word: sentence.split(" ")) {
  30.                 out.collect(new Tuple2<String, Integer>(word, 1));
  31.             }
  32.         }
  33.     }
  35. }
  1. import org.apache.flink.streaming.api.scala._
  2. import org.apache.flink.streaming.api.windowing.time.Time
  4. object WindowWordCount {
  5.   def main(args: Array[String]) {
  7.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  8.     val text = env.socketTextStream("localhost", 9999)
  10.     val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
  11.       .map { (_, 1) }
  12.       .keyBy(0)
  13.       .timeWindow(Time.seconds(5))
  14.       .sum(1)
  16.     counts.print()
  18.     env.execute("Window Stream WordCount")
  19.   }
  20. }

要运行示例程序，首先从终端使用netcat启动输入流：

  1. nc -lk 9999

只需键入一些单词就可以返回一个新单词。这些将是字数统计程序的输入。如果要查看大于1的计数，请在5秒内反复键入相同的单词（如果无法快速输入，则将窗口大小从5秒增加☺）。

数据源

源是您的程序从中读取输入的位置。您可以使用将附加源附加到程序StreamExecutionEnvironment.addSource(sourceFunction)。Flink附带了许多预先实现的源函数，但您可以通过实现SourceFunction非并行源，或通过实现ParallelSourceFunction接口或扩展RichParallelSourceFunction并行源来编写自己的自定义源。

有几个预定义的流源可从以下位置访问StreamExecutionEnvironment：

基于文件的：

  • readTextFile(path)- TextInputFormat逐行读取文本文件，即符合规范的文件，并将它们作为字符串返回。

  • readFile(fileInputFormat, path) -按指定的文件输入格式指定读取（一次）文件。

  • readFile(fileInputFormat, path, watchType, interval,pathFilter, typeInfo) - 这是前两个内部调用的方法。它path根据给定的内容读取文件fileInputFormat。根据提供的内容watchType，此源可以定期监视（每intervalms）新数据（FileProcessingMode.PROCESS_CONTINUOUSLY）的路径，或者处理当前在路径中的数据并退出（FileProcessingMode.PROCESS_ONCE）。使用该pathFilter，用户可以进一步排除正在处理的文件。

实现：

在引擎盖下，Flink将文件读取过程分为两个子任务，即目录监控和数据读取。这些子任务中的每一个都由单独的实体实现。监视由单个非并行（并行性=1）任务实现，而读取由并行运行的多个任务执行。后者的并行性等于工作并行性。单个监视任务的作用是扫描目录（定期或仅一次，具体取决于watchType），找到要处理的文件，将它们分成分割，并将这些拆分分配给下游读卡器。读者是那些将阅读实际数据的人。每个分割仅由一个读取器读取，而读取器可以逐个读取多个分割。

重要笔记：

  • 如果watchType设置为FileProcessingMode.PROCESS_CONTINUOUSLY，则在修改文件时，将完全重新处理其内容。这可以打破“完全一次”的语义，因为在文件末尾追加数据将导致其所有内容被重新处理。

  • 如果watchType设置为FileProcessingMode.PROCESS_ONCE，则源扫描路径一次并退出，而不等待读者完成读取文件内容。当然读者将继续阅读，直到读取所有文件内容。在该点之后关闭源将导致不再有检查点。这可能会导致节点发生故障后恢复速度变慢，因为作业将从上一个检查点恢复读取。

插座为基础的：

  • socketTextStream -从套接字读取。数据元可以用分隔符分隔。基于集合：

  • fromCollection(Collection) - 从JavaJava.util.Collection创建数据流。集合中的所有数据元必须属于同一类型。

  • fromCollection(Iterator, Class) -从迭代器创建数据流。该类指定迭代器返回的数据元的数据类型。

  • fromElements(T …) -从给定的对象序列创建数据流。所有对象必须属于同一类型。

  • fromParallelCollection(SplittableIterator,Class) - 并行地从迭代器创建数据流。该类指定迭代器返回的数据元的数据类型。

  • generateSequence(from, to) -并行生成给定间隔中的数字序列。

自定义：

  • addSource - 附加新的源函数。例如，要从Apache Kafka中读取，您可以使用addSource(newFlinkKafkaConsumer08<>(…))。请参阅连接器以获取更多详

源是您的程序从中读取输入的位置。您可以使用将附加源附加到程序StreamExecutionEnvironment.addSource(sourceFunction)。Flink附带了许多预先实现的源函数，但您可以通过实现SourceFunction非并行源，或通过实现ParallelSourceFunction接口或扩展RichParallelSourceFunction并行源来编写自己的自定义源。

有几个预定义的流源可从以下位置访问StreamExecutionEnvironment：

基于文件的：

  • readTextFile(path)- TextInputFormat逐行读取文本文件，即符合规范的文件，并将它们作为字符串返回。

  • readFile(fileInputFormat, path) -按指定的文件输入格式指定读取（一次）文件。

  • readFile(fileInputFormat, path, watchType, interval,pathFilter) - 这是前两个内部调用的方法。它path根据给定的内容读取文件fileInputFormat。根据提供的内容watchType，此源可以定期监视（每intervalms）新数据（FileProcessingMode.PROCESS_CONTINUOUSLY）的路径，或者处理当前在路径中的数据并退出（FileProcessingMode.PROCESS_ONCE）。使用该pathFilter，用户可以进一步排除正在处理的文件。

实现：

在引擎盖下，Flink将文件读取过程分为两个子任务，即目录监控和数据读取。这些子任务中的每一个都由单独的实体实现。监视由单个非并行（并行性=1）任务实现，而读取由并行运行的多个任务执行。后者的并行性等于工作并行性。单个监视任务的作用是扫描目录（定期或仅一次，具体取决于watchType），找到要处理的文件，将它们分成分割，并将这些拆分分配给下游读卡器。读者是那些将阅读实际数据的人。每个分割仅由一个读取器读取，而读取器可以逐个读取多个分割。

重要笔记：

  • 如果watchType设置为FileProcessingMode.PROCESS_CONTINUOUSLY，则在修改文件时，将完全重新处理其内容。这可以打破“完全一次”的语义，因为在文件末尾追加数据将导致其所有内容被重新处理。

  • 如果watchType设置为FileProcessingMode.PROCESS_ONCE，则源扫描路径一次并退出，而不等待读者完成读取文件内容。当然读者将继续阅读，直到读取所有文件内容。在该点之后关闭源将导致不再有检查点。这可能会导致节点发生故障后恢复速度变慢，因为作业将从上一个检查点恢复读取。

插座为基础的：

  • socketTextStream -从套接字读取。元素可以用分隔符分隔。基于集合：

  • fromCollection(Seq) - 从JavaJava.util.Collection创建数据流。集合中的所有元素必须属于同一类型。

  • fromCollection(Iterator) -从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。

  • fromElements(elements: _*) -从给定的对象序列创建数据流。所有对象必须属于同一类型。

  • fromParallelCollection(SplittableIterator) -并行地从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。

  • generateSequence(from, to) -并行生成给定间隔中的数字序列。

自定义：

  • addSource - 附加新的源函数。例如，要从Apache Kafka中读取，您可以使用addSource(newFlinkKafkaConsumer08<>(…))。请参阅连接器以获取更多详

DataStream转换

有关可用流转换的概述，请参阅 算子。

数据接收

数据接收器使用DataStream并将它们转发到文件，套接字，外部系统或打印它们。Flink带有各种内置输出格式，这些格式封装在DataStreams上的 算子操作后面：

  • writeAsText()/ TextOutputFormat-按字符串顺序写入数据元。通过调用每个数据元的toString（）方法获得字符串。

  • writeAsCsv(…)/ CsvOutputFormat-将元组写为逗号分隔值文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString（）方法。

  • print()/ printToErr()- 在标准输出/标准错误流上打印每个数据元的toString（）值。可选地，可以提供前缀（msg），其前缀为输出。这有助于区分不同的打印调用。如果并行度大于1，则输出也将与生成输出的任务的标识符一起添加。

  • writeUsingOutputFormat()/ FileOutputFormat-自定义文件输出的方法和基类。支持自定义对象到字节的转换。

  • writeToSocket -根据a将数据元写入套接字 SerializationSchema

  • addSink -调用自定义接收器函数。Flink捆绑了其他系统（如ApacheKafka）的连接器，这些系统实现为接收器函数。

数据接收器使用DataStream并将它们转发到文件，套接字，外部系统或打印它们。Flink带有各种内置输出格式，这些格式封装在DataStreams上的 算子操作后面：

  • writeAsText()/ TextOutputFormat-按字符串顺序写入元素。通过调用每个元素的toString（）方法获得字符串。

  • writeAsCsv(…)/ CsvOutputFormat-将元组写为逗号分隔值文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString（）方法。

  • print()/ printToErr()- 在标准输出/标准错误流上打印每个元素的toString（）值。可选地，可以提供前缀（msg），其前缀为输出。这有助于区分不同的打印调用。如果并行度大于1，则输出也将与生成输出的任务的标识符一起添加。

  • writeUsingOutputFormat()/ FileOutputFormat-自定义文件输出的方法和基类。支持自定义对象到字节的转换。

  • writeToSocket -根据a将元素写入套接字 SerializationSchema

  • addSink -调用自定义接收器函数。Flink捆绑了其他系统（如ApacheKafka）的连接器，这些系统实现为接收器函数。

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

要将流可靠，准确地一次传送到文件系统，请使用flink-connector-filesystem。此外，通过该.addSink(…)方法的自定义实现可以参与Flink的精确一次语义检查点。

迭代

迭代流程序实现步进函数并将其嵌入到IterativeStream。由于DataStream程序可能永远不会完成，因此没有最大迭代次数。相反，您需要指定流的哪个部分反馈到迭代，哪个部分使用split转换或转发到下游filter。在这里，我们展示了使用过滤器的示例。首先，我们定义一个IterativeStream

  1. IterativeStream<Integer> iteration = input.iterate();

然后，我们使用一系列转换指定将在循环内执行的逻辑（这里是一个简单的map转换）

  1. DataStream<Integer> iterationBody = iteration.map(/* this is executed many times */);

要关闭迭代并定义迭代尾部，请调用closeWith(feedbackStream)方法IterativeStream。赋予closeWith函数的DataStream 将反馈给迭代头。常见的模式是使用过滤器来分离反馈的流的部分和向前传播的流的部分。这些滤波器可以例如定义“终止”逻辑，其中允许元件向下游传播而不是反馈。

  1. iteration.closeWith(iterationBody.filter(/* one part of the stream */));
  2. DataStream<Integer> output = iterationBody.filter(/* some other part of the stream */);

例如，这里是从一系列整数中连续减去1直到它们达到零的程序：

  1. DataStream<Long> someIntegers = env.generateSequence(0, 1000);
  3. IterativeStream<Long> iteration = someIntegers.iterate();
  5. DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
  6.   @Override
  7.   public Long map(Long value) throws Exception {
  8.     return value - 1 ;
  9.   }
  10. });
  12. DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
  13.   @Override
  14.   public boolean filter(Long value) throws Exception {
  15.     return (value > 0);
  16.   }
  17. });
  19. iteration.closeWith(stillGreaterThanZero);
  21. DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
  22.   @Override
  23.   public boolean filter(Long value) throws Exception {
  24.     return (value <= 0);
  25.   }
  26. });

迭代流程序实现步进函数并将其嵌入到IterativeStream。由于DataStream程序可能永远不会完成，因此没有最大迭代次数。相反，您需要指定流的哪个部分反馈到迭代，哪个部分使用split转换或转发到下游filter。这里，我们展示了一个示例迭代，其中正文（重复的计算部分）是一个简单的映射转换，反馈的元素由使用过滤器向下游转发的元素区分。

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

例如，这里是从一系列整数中连续减去1直到它们达到零的程序：

  1. val someIntegers: DataStream[Long] = env.generateSequence(0, 1000)
  3. val iteratedStream = someIntegers.iterate(
  4.   iteration => {
  5.     val minusOne = iteration.map( v => v - 1)
  6.     val stillGreaterThanZero = minusOne.filter (_ > 0)
  7.     val lessThanZero = minusOne.filter(_ <= 0)
  8.     (stillGreaterThanZero, lessThanZero)
  9.   }
  10. )

执行参数

该StreamExecutionEnvironment包含ExecutionConfig允许为运行时设置工作的具体配置值。

有关大多数参数的说明，请参阅执行配置。这些参数特别适用于DataStream API：

  • setAutoWatermarkInterval(long milliseconds)：设置自动水印发射的间隔。您可以使用获取当前值long getAutoWatermarkInterval()

容错

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

控制延迟

默认情况下，数据元不会逐个传输到网络上（这会导致不必要的网络流量），但会被缓冲。可以在Flink配置文件中设置缓冲区的大小（实际在计算机之间传输）。虽然此方法适用于优化吞吐量，但当传入流速度不够快时，可能会导致延迟问题。要控制吞吐量和延迟，您可以env.setBufferTimeout(timeoutMillis)在运行环境（或单个 算子）上使用以设置缓冲区填充的最长等待时间。在此之后，即使缓冲区未满，也会自动发送缓冲区。此超时的默认值为100毫秒。

用法：

  1. LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  2. env.setBufferTimeout(timeoutMillis);
  4. env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);
  1. val env: LocalStreamEnvironment = StreamExecutionEnvironment.createLocalEnvironment
  2. env.setBufferTimeout(timeoutMillis)
  4. env.generateSequence(1,10).map(myMap).setBufferTimeout(timeoutMillis)

为了最大化吞吐量，设置setBufferTimeout(-1)哪个将删除超时和缓冲区只有在它们已满时才会被刷新。要最小化延迟，请将超时设置为接近0的值（例如5或10 ms）。应避免缓冲区超时为0，因为它可能导致严重的性能下降。

调试

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

Flink通过支持IDE内的本地调试，测试数据的注入和结果数据的收集，提供了显着简化数据分析程序开发过程的函数。本节提供了一些如何简化Flink程序开发的提示。

本地运行环境

A LocalStreamEnvironment在创建它的同一JVM进程中启动Flink系统。如果从IDE启动LocalEnvironment，则可以在代码中设置断点并轻松调试程序。

创建LocalEnvironment并使用如下：

  1. final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  3. DataStream<String> lines = env.addSource(/* some source */);
  4. // build your program
  6. env.execute();
  1. val env = StreamExecutionEnvironment.createLocalEnvironment()
  3. val lines = env.addSource(/* some source */)
  4. // build your program
  6. env.execute()

收集数据源

Flink提供了特殊的数据源，这些数据源由Java集合支持，以方便测试。一旦程序经过测试，源和接收器可以很容易地被读取/写入外部系统的源和接收器替换。

集合数据源可以使用如下：

  1. final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  3. // Create a DataStream from a list of elements
  4. DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);
  6. // Create a DataStream from any Java collection
  7. List<Tuple2<String, Integer>> data = ...
  8. DataStream<Tuple2<String, Integer>> myTuples = env.fromCollection(data);
  10. // Create a DataStream from an Iterator
  11. Iterator<Long> longIt = ...
  12. DataStream<Long> myLongs = env.fromCollection(longIt, Long.class);
  1. val env = StreamExecutionEnvironment.createLocalEnvironment()
  3. // Create a DataStream from a list of elements
  4. val myInts = env.fromElements(1, 2, 3, 4, 5)
  6. // Create a DataStream from any Collection
  7. val data: Seq[(String, Int)] = ...
  8. val myTuples = env.fromCollection(data)
  10. // Create a DataStream from an Iterator
  11. val longIt: Iterator[Long] = ...
  12. val myLongs = env.fromCollection(longIt)

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

迭代器数据接收器

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

  1. import org.apache.flink.streaming.experimental.DataStreamUtils
  3. DataStream<Tuple2<String, Integer>> myResult = ...
  4. Iterator<Tuple2<String, Integer>> myOutput = DataStreamUtils.collect(myResult)
  1. import org.apache.flink.streaming.experimental.DataStreamUtils
  2. import scala.collection.JavaConverters.asScalaIteratorConverter
  4. val myResult: DataStream[(String, Int)] = ...
  5. val myOutput: Iterator[(String, Int)] = DataStreamUtils.collect(myResult.javaStream).asScala

注意： flink-streaming-contrib模块已从Flink1.5.0中删除。它的课程已被移入flink-streaming-java和flink-streaming-scala。

下一步