Flink实战(五) - DataStream API编程（上）

2021-12-07 353

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

简介： Flink实战(五) - DataStream API编程（上）

1 概述

Flink中的DataStream程序是实现数据流转换的常规程序（例如，过滤，更新状态，定义窗口，聚合）。

最初从各种源（例如，消息队列，套接字流，文件）创建数据流。

结果通过接收器返回，接收器可以例如将数据写入文件或标准输出（例如命令行终端）。

Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在许多计算机的集群上执行。

有关Flink API基本概念的介绍，请参阅

https://blog.csdn.net/qq_33589510/article/details/89893394

2 入门案例

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

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
public class WindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);
        dataStream.print();
        env.execute("Window WordCount");
    }
    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

nc -lk 9999

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

Socket输入
程序输出

创建一个新数据流，其中包含从套接字无限接收的字符串。接收的字符串由系统的默认字符集解码，使用“\ n”作为分隔符。当socket关闭时，阅读器立即终止。

Scala版本

3 Data source

源是您的程序从中读取输入的位置。可以使用

StreamExecutionEnvironment.addSource（sourceFunction）

将源附加到程序

Flink附带了许多预置实现的源函数，但你可以通过为非并行源实现SourceFunction，或者通过实现ParallelSourceFunction接口或为并行源扩展RichParallelSourceFunction来编写自己的自定义源。

可以从StreamExecutionEnvironment访问几个预定义的流源：

3.1 基于文件

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），找到要处理的文件，将它们分层分割，并将这些拆分分配给下游读卡器。读者是那些将阅读实际数据的人。每个分割仅由一个读取器读取，而读取器可以逐个读取多个分割。

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

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