1、Flink流处理（DataStream API）概览（Data Source、Data Sink）

秒速五厘米 2021-10-19 07:52 399阅读 0赞

Flink中的DataStream程序是在数据流上实现转换的常规程序(例如，filtering, updating state, defining windows, aggregating)。数据流最初是由不同的源创建的(例如，message queues, socket streams, files)。结果通过接收器返回，例如，接收器可以将数据写入文件或标准输出(例如the command line terminal)。Flink程序在各种上下文中运行，独立运行或嵌入到其他程序中。执行可以在本地JVM中进行，也可以在许多机器的集群上进行。

有关Flink API的基本概念的介绍，请参阅[Flink基础之API，DataSet、DataStream、批、流][Flink_API_DataSet_DataStream]

以下为简易的Flink DataStream程序。

### 样例程序 ###

下面的程序是一个完整的、可工作的流窗口单词计数应用程序示例，它在5秒内计算来自web套接字的单词。您可以复制并粘贴代码以在本地运行它。

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.utils.ParameterTool;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
    import org.junit.Test;
    
    public class CommonDemo1Test {
        @Test
        public void demo1() throws Exception {
            ParameterTool params = ParameterTool.fromArgs(new String[0]);
            StreamExecutionEnvironment ev = StreamExecutionEnvironment.getExecutionEnvironment();
            ev.getConfig().setGlobalJobParameters(params);
            DataStream<String> text = null;
            if (params.has("input")) {
                text = ev.readTextFile(params.get("input"));
            } else {
                System.out.println("Executing WordCount example with default input data set.");
                System.out.println("Use --input to specify file input.");
                text = ev.fromElements(WordCountData.WORDS);
            }
            System.out.println(text);
            SingleOutputStreamOperator<Tuple2<String, Integer>> res = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                    String[] tokens = s.toLowerCase().split("\\W+");
                    for (String token : tokens) {
                        if (token.length() > 1) {
                            collector.collect(Tuple2.of(token, 1));
                        }
                    }
                }
            }).keyBy(0).sum(1);
    
    
            if (params.has("output")) {
                res.writeAsText(params.get("output"));
            } else {
                System.out.println("Printing result to stdout. Use --output to specify output path.");
                res.print();
            }
    
            // execute program
            ev.execute("Streaming WordCount");
        }
    
        @Test
        public void demo2() {
    
        }
    }

public class WordCountData {
        public static final String[] WORDS = new String[]{
                "To be, or not to be,--that is the question:--",
                "Whether 'tis nobler in the mind to suffer",
                "The slings and arrows of outrageous fortune",
                "Or to take arms against a sea of troubles,",
                "And by opposing end them?--To die,--to sleep,--",
                "No more; and by a sleep to say we end",
                "The heartache, and the thousand natural shocks",
                "That flesh is heir to,--'tis a consummation",
                "Devoutly to be wish'd. To die,--to sleep;--",
                "To sleep! perchance to dream:--ay, there's the rub;",
                "For in that sleep of death what dreams may come,",
                "When we have shuffled off this mortal coil,",
                "Must give us pause: there's the respect",
                "That makes calamity of so long life;",
                "For who would bear the whips and scorns of time,",
                "The oppressor's wrong, the proud man's contumely,",
                "The pangs of despis'd love, the law's delay,",
                "The insolence of office, and the spurns",
                "That patient merit of the unworthy takes,",
                "When he himself might his quietus make",
                "With a bare bodkin? who would these fardels bear,",
                "To grunt and sweat under a weary life,",
                "But that the dread of something after death,--",
                "The undiscover'd country, from whose bourn",
                "No traveller returns,--puzzles the will,",
                "And makes us rather bear those ills we have",
                "Than fly to others that we know not of?",
                "Thus conscience does make cowards of us all;",
                "And thus the native hue of resolution",
                "Is sicklied o'er with the pale cast of thought;",
                "And enterprises of great pith and moment,",
                "With this regard, their currents turn awry,",
                "And lose the name of action.--Soft you now!",
                "The fair Ophelia!--Nymph, in thy orisons",
                "Be all my sins remember'd."
        };
    }

###  Data Sources（数据源） ###

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

有几个预定义的流资源可以从StreamExecutionEnvironment使用：

*  File

1.  readTextFile(path)：按照TextInputFormat规范逐行读取并以字符串形式返回的文件。
2.  readFile(fileInputFormat, path)：按照指定的文件输入格式读取(一次)文件。
3.  readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)：这是前两个方法在内部调用的方法。它根据给定的fileInputFormat读取路径中的文件。根据所提供的watchType，此源可以定期(每隔一段时间)监视新数据的路径(`FileProcessingMode.PROCESS_CONTINUOUSLY)`，或者一次性处理当前路径中的数据并退出(`FileProcessingMode.PROCESS_ONCE)`。使用路径过滤器，用户可以进一步排除正在处理的文件。

**拆分：**

在底层，Flink将文件读取过程分成两个子任务，即目录监视和数据读取。每个子任务都由一个单独的实体实现。监视由单个非并行(parallelism = 1)任务实现，而读取由多个并行运行的任务执行。后者的并行性等于作业并行性。单个监视任务的作用是扫描目录(定期或仅扫描一次，这取决于watchType)，找到要处理的文件，将它们分成几部分，并将这些部分分配给下游的读取器。读取器将读取实际数据。每个数据块只能由一个读取器读取，而一个读取器可以逐个读取多个拆分。

**注意：**

1.   如果watchType设置为`FileProcessingMode.PROCESS_CONTINUOUSLY`，当一个文件被修改时，它的内容被完全重新处理。这可能会打破 “exactly-once” 语义，因为在文件末尾附加数据将导致重新处理所有内容。
2.  如果watchType设置为`FileProcessingMode.PROCESS_ONCE`，源程序扫描路径一次并退出，而不等待读取器完成文件内容的读取。当然，readers将继续阅读，直到所有的文件内容被读取。关闭源将导致在该点之后不再有checkpoints。这可能导致节点故障后恢复较慢，因为作业将从最后一个检查点恢复读取。

*  Socket

1.  socketTextStream：从网络读取，元素可以用分隔符分隔。

*  Collection

1.  fromCollection(collection)：从Java Java.util. Collection创建数据流。集合中的所有元素必须具有相同的类型。
2.  fromCollection(Iterator, Class)：从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
3.  fromElement(T ...)：从给定的对象序列创建数据流。所有对象必须具有相同的类型。
4.  fromParallelCollectrion(SplitTableIterator, Class)：并行地从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
5.  generateSequence(from, to)：并行地生成给定区间内的数字序列。

*  自定义

1.  addSource：附加一个新的源函数。例如，要从Apache Kafka读取数据，可以使用addSource(new FlinkKafkaConsumer08<>(…))。

### DataStream数据转换 ###

主要为operators，请查阅[Flink流处理（Stream API）- Operators（操作数据流）][Flink_Stream API_- Operators]

### Data Sinks ###

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

1.  `writeAsText()` / `TextOutputFormat：`将元素按行写入字符串。通过调用每个元素的toString()方法获得字符串。
2.  `writeAsCsv(...)` / `CsvOutputFormat：`将元组写入逗号分隔的值文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。
3.  `print()` / `printToErr()：`在标准输出/标准错误流上打印每个元素的toString()值。此外，还可以提供前缀(msg)作为输出的前缀。这有助于区分不同的打印调用。如果并行度大于1，输出也将以生成输出的任务的标识符作为前缀。
4.  `writeUsingOutputFormat()` / `FileOutputFormat：`方法和自定义文件输出的基类。支持自定义对象到字节的转换。
5.  `writeToSocket：`根据SerializationSchema将元素写入套接字
6.  `addSink：`调用自定义接收器函数。Flink与其他系统(如Apache Kafka)的连接器捆绑在一起，这些连接器作为接收器函数实现。

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

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

## Iterations（遍历） ##

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

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

然后，我们指定将在循环中使用一系列转换执行的逻辑(这里是一个简单的映射转换)

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

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

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

例如，这里有一个程序，它不断地从一系列整数中减去1，直到它们达到零：

DataStream<Long> someIntegers = env.generateSequence(0, 1000);
    
    IterativeStream<Long> iteration = someIntegers.iterate();
    
    DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
      @Override
      public Long map(Long value) throws Exception {
        return value - 1 ;
      }
    });
    
    DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
      @Override
      public boolean filter(Long value) throws Exception {
        return (value > 0);
      }
    });
    
    iteration.closeWith(stillGreaterThanZero);
    
    DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
      @Override
      public boolean filter(Long value) throws Exception {
        return (value <= 0);
      }
    });

### 执行参数配置 ###

StreamExecutionEnvironment包含ExecutionConfig，它允许为运行时设置特定于作业的配置值。

//TODO后续详细描述

**容错性**

State和Checkpointing 请查阅[Flink流处理（Stream API）- State & Fault Tolerance（状态和容错）之 Checkpointing][Flink_Stream API_- State _ Fault Tolerance_ Checkpointing]

**控制延迟**

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

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
    env.setBufferTimeout(timeoutMillis);
    
    env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);

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

### 调试以排除故障 ###

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

通过支持IDE中的本地调试、测试数据的注入和结果数据的收集，Flink提供了一些特性，可以显著简化数据分析程序的开发过程。

**本地开发环境**

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

本地环境的创建和使用如下：

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

**Collection Data Sources（收集数据源）**

Flink提供了由Java集合支持的特殊数据源，以简化测试。一旦程序经过测试，源和接收器就可以很容易地替换为从外部系统读写的源和接收器。

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

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
    
    // Create a DataStream from a list of elements
    DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);
    
    // Create a DataStream from any Java collection
    List<Tuple2<String, Integer>> data = ...
    DataStream<Tuple2<String, Integer>> myTuples = env.fromCollection(data);
    
    // Create a DataStream from an Iterator
    Iterator<Long> longIt = ...
    DataStream<Long> myLongs = env.fromCollection(longIt, Long.class);

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

**Iterator Data Sink（迭代器数据收集器）**

Flink还提供了一个接收器来收集数据流结果，用于测试和调试。它的用途如下：

import org.apache.flink.streaming.experimental.DataStreamUtils
    
    DataStream<Tuple2<String, Integer>> myResult = ...
    Iterator<Tuple2<String, Integer>> myOutput = DataStreamUtils.collect(myResult)

**注意：**从Flink 1.5.0中删除了Flink -stream -contrib模块。它的类已经迁移到flink-streaming-java和flink-streaming-scala中。

[Flink_API_DataSet_DataStream]: https://blog.csdn.net/zpf_940810653842/article/details/99636978
[Flink_Stream API_- Operators]: https://blog.csdn.net/zpf_940810653842/article/details/99751920
[Flink_Stream API_- State _ Fault Tolerance_ Checkpointing]: https://blog.csdn.net/zpf_940810653842/article/details/99817128