Flink DataStream API编程指南

- 日理万妓 2023-02-15 07:58 50阅读 0赞

## 一.简介 ##

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

## 二.代码案例 ##

以下程序是流式窗口单词计数应用程序的一个完整的工作示例，该应用程序在5秒的窗口中对来自Socket套接字的单词进行计数。

package cn.socket
    
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.time.Time
    
    // 数据类型异常，动态数据引入
    // import org.apache.flink.api.scala._ // 数据类型异常，静态数据引入
    
    /** * Created by Administrator on 2020/3/22. */
    object SocketWindowWordCount { 
      def main(args: Array[String]) : Unit = { 
    
        // 指定的IP和接口
        val hostname: String = "192.168.136.7"
        val port: Int = 9999
    
        // 获取流处理环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 获取socket计算数据
        val windowCounts = env.socketTextStream(hostname, port, '\n')
          .flatMap {  w => w.split("\\s") }
          .map(w => (w, 1))
          .keyBy(_._1)
          .timeWindow(Time.seconds(5))
          .sum(1)
    
        // 设置并行度，打印
        windowCounts.print().setParallelism(1)
    
        // 执行
        env.execute("Socket Window WordCount")
      }
    }

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

nc -lk 9999

只需输入一些单词【以空格进行分隔】，然后按回车键即可获得一个新单词串。这些将作为单词计数程序的输入。  
![在这里插入图片描述][20200604195121454.png]  
![在这里插入图片描述][20200604195140860.png]

## 三.数据源 ##

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

可以使用StreamExecutionEnvironment在以下位置访问几个预定义的流源：  
1.基于文件：

*  readTextFile(path)：TextInputFormat逐行读取文本文件，符合规范的文件，将其作为字符串返回。
 *  readFile(fileInputFormat, path) ：根据指定的文件输入格式读取（一次）文件。
 *  readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)：这是前两个内部调用的方法。path根据给定的fileInputFormat类型读取文件。根据提供的内容watchType，此源可以定期（每interval ms）监视路径中的新数据。  
    FileProcessingMode.PROCESS\_CONTINUOUSLY、FileProcessingMode.PROCESS\_ONCE，持续或处理一次路径中当前的数据并退出。使用pathFilter，用户可以进一步从文件夹中排除文件。  
    **实施：**  
    在后台，Flink将文件读取过程分为两个子任务，即目录监视和数据读取。这些子任务中的每一个都是由单独的实体实现的。监视由单个非并行任务实现，而读取由并行运行的多个任务执行。后者的并行性等于作业并行性。单个监视任务的作用是扫描目录（根据定期扫描或仅扫描一次watchType），找到要处理的文件，将它们分成多个部分，并将这些拆分分配给下游reader。reader将read实际数据。每个拆分只能由一个reader读取，而reader可以read多个拆分。  
    **重要笔记：**  
    如果将watchType设置为FileProcessingMode.PROCESS\_CONTINUOUSLY，则在修改文件时，将完全重新处理其内容。这可能会破坏“完全一次”的语义，因为在文件末尾附加数据将导致重新处理其所有内容。  
    如果将watchType设置为FileProcessingMode.PROCESS\_ONCE，则源将扫描路径一次并退出，而无需等待读取器完成文件内容的读取。当然，读取器将继续读取，直到读取了所有文件内容。关闭源将导致在该点之后没有更多检查点。这可能导致节点故障后恢复速度变慢，因为作业将从上一个检查点恢复读取。

2.基于套接字

*  socketTextStream-从套接字读取。元素可以由定界符分隔。

3.基于集合

*  fromCollection(Collection)：从Java的java.util.Collection创建数据流。集合中的所有元素必须具有相同的类型。
 *  fromCollection(Iterator, Class)：从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
 *  fromElements(T …)：从给定的对象序列创建数据流。所有对象必须具有相同的类型。
 *  fromParallelCollection(SplittableIterator, Class)：从迭代器并行创建数据流。该类指定迭代器返回的元素的数据类型。
 *  generateSequence(from, to) ：并行生成给定间隔中的数字序列。

4.自定义

*  addSource：附加新的源功能。可以使用 addSource(new FlinkKafkaConsumer08<>(…))。

## 四.DataStream转换 ##

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

*  writeAsText()/ TextOutputFormat：将元素按行写为字符串。通过调用每个元素的toString()方法获得字符串。
 *  writeAsCsv(…)/ CsvOutputFormat：将元组写为逗号分隔的值文件。行和字段定界符是可配置的。每个字段的值来自对象的toString()方法。
 *  print()/ printToErr() ：在标准输出/标准错误流上打印每个元素的toString()值。可选地，可以提供前缀（msg），该前缀在输出之前。这可以帮助区分不同的打印调用。如果并行度大于1，则输出之前还将带有产生输出的任务的标识符。
 *  writeUsingOutputFormat()/ FileOutputFormat：自定义文件输出的方法和基类。支持自定义对象到字节的转换。
 *  writeToSocket：根据SerializationSchema将元素写入套接字。
 *  addSink：调用自定义接收器功能。Flink捆绑有连接到其他系统（例如Apache Kafka）的连接器，这些连接器实现为接收器功能。

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

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

## 五.迭代流程序 ##

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

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

然后，我们使用一系列转换（此处为简单map转换）指定将在循环内执行的逻辑：

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

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

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允许用于运行组工作的具体配置值。  
例如，设置自动水印：

setAutoWatermarkInterval(long milliseconds)
    long getAutoWatermarkInterval()

## 七.控制延迟 ##

默认情况下，元素不会在网络上一对一传输（这会导致不必要的网络通信），但是会进行缓冲。缓冲区的大小（实际上是在计算机之间传输的）可以在Flink配置文件中设置。尽管此方法可以优化吞吐量，但是当传入流不够快时，它可能会导致延迟问题。为了控制吞吐量和延迟，您可以env.setBufferTimeout(timeoutMillis)在执行环境（或各个运算符）上使用来设置缓冲区填充的最大等待时间。在此时间之后，即使缓冲区未满，也会自动发送缓冲区。此超时的默认值为100毫秒。

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

为了最大化吞吐量，请设置set setBufferTimeout(-1)来消除超时，并且仅在缓冲区已满时才刷新它们。为了使延迟最小化，请将超时设置为接近0的值（例如5或10 ms）。应避免将缓冲区超时设置为0，因为它可能导致严重的性能下降。

[20200604195121454.png]: https://img-blog.csdnimg.cn/20200604195121454.png
[20200604195140860.png]: https://img-blog.csdnimg.cn/20200604195140860.png