Flink初入门-API使用(补)

偏执的太偏执、 2021-12-20 06:11 220阅读 0赞

**目录**

数据存储/接收器

迭代：

执行参数：

容错：

控制延迟：

--------------------

## 数据存储/接收器 ##

数据接收器使用DataStream将他们转发到文件，socket，外部系统或者打印他们。Flink带有各种被指的输出格式，这些格式封装再DataStream上的算子操作后面：

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

> writeAsCsv(...) / CsvOutputFormat - 将元组写为逗号分割的文件。(Execl/hive等使用)。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。 初学时记得调用execute方法后才会有程序输出。

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

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

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

> addSink --调用自定义接收器函数。Flink捆绑了其它系统(如Apache Kafka)的连接器，这些系统实现为接收器函数。

请注意：

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

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

## 迭代： ##

迭代流程序实现不仅函数并将其嵌入到IterativeStream。由于DataStream程序可能永远不会完成，因此没有最大迭代次数。相反，您需要指定流的哪个部分反馈到迭代，那个部分使用split转行或转发到下游fliter。在这里，我们展示了使用过滤器的示例。首先，我们定义一个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 stream*/));
    DataStream<Integer> output = iterationBody.filter(/* some other part of the stream */);

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

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

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

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

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

val someIntegers:DataStream[Long] = env.generateSequence(0,1000)
    val iteratedStream = someIntegers.iterate(
        iteration => {
            val minusOne = iteration.map( v => v -1)
            val stillGreaterThanZero = minusOne.filter (_ > 0)
            val lessThanZero = minusOne.fliter(_ <= 0)
            (stillGreaterThanZero,lessThanZero)
            }
    )

## 执行参数： ##

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

有关大多数参数的说明，请参阅[执行配置][Link 1]。这些参数特别适合DataStream API：

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

## 容错： ##

参阅-[容错处理][Link 2]

## 控制延迟：    ##

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

val env:LocalStreamEnvironment = StreamExecutionEnvironmet.createLocalEnvironment
    env.setBufferTimeout(timeoutMillis)
    
    env.generateSequence(1,10).map(myMap).setBufferTimeout(timeoutMillis)

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

[Link 1]: https://flink.sojb.cn/dev/execution_configuration.html
[Link 2]: https://blog.csdn.net/xiaoxin1024/article/details/94208210