Flink (五) ：DataStream API （二）

🕗 发布于 2025-01-16 19:22 flink 大数据

1. Transformations

用户通过算子能将一个或多个 DataStream 转换成新的 DataStream，在应用程序中可以将多个数据转换算子合并成一个复杂的数据流拓扑。

1.1 Map

DataStream → DataStream: 输入一个元素同时输出一个元素。下面是将输入流中元素数值加倍的 map function：

DataStream<Integer> dataStream = //...
dataStream.map(new MapFunction<Integer, Integer>() {
    @Override
    public Integer map(Integer value) throws Exception {
        return 2 * value;
    }
});

1.2 FlatMap

DataStream → DataStream: 输入一个元素同时产生零个、一个或多个元素。下面是将句子拆分为单词的 flatmap function：

dataStream.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String value, Collector<String> out)
        throws Exception {
        for(String word: value.split(" ")){
            out.collect(word);
        }
    }
});

1.3 Filter

DataStream → DataStream: 为每个元素执行一个布尔 function，并保留那些 function 输出值为 true 的元素。下面是过滤掉零值的 filter：

dataStream.filter(new FilterFunction<Integer>() {
    @Override
    public boolean filter(Integer value) throws Exception {
        return value != 0;
    }
});

1.4 KeyBy

DataStream → KeyedStream: 在逻辑上将流划分为不相交的分区。具有相同 key 的记录都分配到同一个分区。在内部， keyBy() 是通过哈希分区实现的。

dataStream.keyBy(value -> value.getSomeKey());
dataStream.keyBy(value -> value.f0);

以下情况，一个类不能作为 key：

它是一种 POJO 类，但没有重写 hashCode() 方法而是依赖于 Object.hashCode() 实现。
它是任意类的数组。

1.5 Reduce

KeyedStream → DataStream: 在相同 key 的数据流上“滚动”执行 reduce。将当前元素与最后一次 reduce 得到的值组合然后输出新值。

下面是创建局部求和流的 reduce function：

keyedStream.reduce(new ReduceFunction<Integer>() {
    @Override
    public Integer reduce(Integer value1, Integer value2)
    throws Exception {
        return value1 + value2;
    }
});

1.6 Union

DataStream* → DataStream: 将两个或多个数据流联合来创建一个包含所有流中数据的新流。注意：如果一个数据流和自身进行联合，这个流中的每个数据将在合并后的流中出现两次。同时注意合并的流必须是相同类型才可以合并

dataStream.union(otherStream1, otherStream2, ...);

1.7 Connect

DataStream,DataStream → ConnectedStream: “连接” 两个数据流并保留各自的类型。connect 允许在两个流的处理逻辑之间共享状态。

DataStream<Integer> someStream = //...
DataStream<String> otherStream = //...

ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);

1.8 CoMap, CoFlatMap

ConnectedStream → DataStream: 类似于在连接的数据流上进行 map 和 flatMap。

connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {
    @Override
    public Boolean map1(Integer value) {
        return true;
    }

    @Override
    public Boolean map2(String value) {
        return false;
    }
});
connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {

   @Override
   public void flatMap1(Integer value, Collector<String> out) {
       out.collect(value.toString());
   }

   @Override
   public void flatMap2(String value, Collector<String> out) {
       for (String word: value.split(" ")) {
         out.collect(word);
       }
   }
});

1.9 Cache

DataStream → CachedDataStream: 把算子的结果缓存起来。目前只支持批执行模式下运行的作业。算子的结果在算子第一次执行的时候会被缓存起来，之后的作业中会复用该算子缓存的结果。如果算子的结果丢失了，它会被原来的算子重新计算并缓存。

DataStream<Integer> dataStream = //...
CachedDataStream<Integer> cachedDataStream = dataStream.cache();
cachedDataStream.print(); // Do anything with the cachedDataStream
...
env.execute(); // Execute and create cache.
        
cachedDataStream.print(); // Consume cached result.
env.execute();

2. 旁路输出

除了由 DataStream 操作产生的主要流之外，你还可以产生任意数量的旁路输出结果流。结果流中的数据类型不必与主要流中的数据类型相匹配，并且不同旁路输出的类型也可以不同。当你需要拆分数据流时，通常必须复制该数据流，然后从每个流中过滤掉不需要的数据，这个操作十分有用。

使用旁路输出时，首先需要定义用于标识旁路输出流的 OutputTag：

可以通过以下方法将数据发送到旁路输出：

ProcessFunction
KeyedProcessFunction
CoProcessFunction
KeyedCoProcessFunction
ProcessWindowFunction
ProcessAllWindowFunction

你可以使用在上述方法中向用户暴露的 Context 参数，将数据发送到由 OutputTag 标识的旁路输出。这是从 ProcessFunction 发送数据到旁路输出的示例：

DataStream<Integer> input = ...;

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = input
  .process(new ProcessFunction<Integer, Integer>() {

      @Override
      public void processElement(
          Integer value,
          Context ctx,
          Collector<Integer> out) throws Exception {
        // 发送数据到主要的输出
        out.collect(value);

        // 发送数据到旁路输出
        ctx.output(outputTag, "sideout-" + String.valueOf(value));
      }
    });

你可以在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流。这将产生一个与旁路输出流结果类型一致的 DataStream：

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = ...;

DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);

3. Data Sinks

Data sinks 使用 DataStream 并将它们转发到文件、套接字、外部系统或打印它们。Flink 自带了多种内置的输出格式，这些格式相关的实现封装在 DataStreams 的算子里：

print() / printToErr() - 在标准输出/标准错误流上打印每个元素的 toString() 值。可选地，可以提供一个前缀（msg）附加到输出。这有助于区分不同的 print 调用。如果并行度大于1，输出结果将附带输出任务标识符的前缀。
writeToSocket - 根据 SerializationSchema 将元素写入套接字。
addSink - 调用自定义 sink function。Flink 捆绑了连接到其他系统（例如 Apache Kafka）的连接器，这些连接器被实现为 sink functions。

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

为了将流可靠地、精准一次地传输到文件系统中，请使用 FileSink。此外，通过 .addSink(...) 方法调用的自定义实现也可以参与 Flink 的 checkpointing，以实现精准一次的语义。

4. 任务触发

一旦指定了完整的程序，需要调用 StreamExecutionEnvironment 的 execute() 方法来触发程序执行。根据 ExecutionEnvironment 的类型，执行会在你的本地机器上触发，或将你的程序提交到某个集群上执行。

execute() 方法将等待作业完成，然后返回一个 JobExecutionResult，其中包含执行时间和累加器结果。

如果不想等待作业完成，可以通过调用 StreamExecutionEnvironment 的 executeAsync() 方法来触发作业异步执行。它会返回一个 JobClient，你可以通过它与刚刚提交的作业进行通信。如下是使用 executeAsync() 实现 execute() 语义的示例。

final JobClient jobClient = env.executeAsync();

final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult().get();

关于程序执行的最后一部分对于理解何时以及如何执行 Flink 算子是至关重要的。所有 Flink 程序都是延迟执行的：当程序的 main 方法被执行时，数据加载和转换不会直接发生。相反，每个算子都被创建并添加到 dataflow 形成的有向图。当执行被执行环境的 execute() 方法显示地触发时，这些算子才会真正执行。程序是在本地执行还是在集群上执行取决于执行环境的类型。

延迟计算允许你构建复杂的程序，Flink 会将其作为一个整体的计划单元来执行。

5. 控制延迟

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

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
env.setBufferTimeout(timeoutMillis);

env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);

为了最大限度地提高吞吐量，设置 setBufferTimeout(-1) 来删除超时，这样缓冲区仅在它们已满时才会被刷新。要最小化延迟，请将超时设置为接近 0 的值（例如 5 或 10 毫秒）。应避免超时为 0 的缓冲区，因为它会导致严重的性能下降。

原文地址：https://blog.csdn.net/weixin_41914554/article/details/145117352

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Autodl转发端口，在本地机器上运行Autodl服务器中的ipynb文件
下一篇：虚幻基础0

使用 Python 操作 MySQL 数据库
MySQL 是一种关系型数据库管理系统 (RDBMS)，基于 SQL（结构化查询语言）进行操作。它使用表格来存储数据，并支持事务、外键、索引等功能。Python 还提供了 ORM（对象关系映射）库，用
阅读更多2025-01-16
文献阅读分享：DKN（WWW2018） - 深度知识感知网络在新闻推荐中的应用
DKN的核心是知识感知卷积神经网络（KCNN）和注意力机制。KCNN将新闻的词嵌入、实体嵌入和上下文实体嵌入作为多通道输入，通过卷积操作提取新闻的特征表示。具体来说，KCNN首先将新闻标题中的每个词与
阅读更多2025-01-16
vue router的使用
📌。
阅读更多2025-01-16
数据库第四天作业
INSERT INTO `worker` (`部门号`, `职工号`, `工作时间`, `工资`, `政治面貌`, `姓名`, `出生日期`) VALUES (102, 1004, '2016-10-
阅读更多2025-01-16
SpringMVC复习笔记
是基于构建的原始Web框架，从一开始就包含在中。正式名称来自其源模块的名称（），但它通常被称为前后端不分离: controller 通过共享域对象，让动态页面从共享域取数据然后返回装配好的 html
阅读更多2025-01-16
【网络协议】ACL（访问控制列表）第二部分
例如，如果我们配置了 ACL 来阻止主机 A 的 Web 流量，而允许主机 B 的 Web 流量，那么如果主机 A 无法访问 Web，而主机 B 能够正常访问，说明 ACL 配置是正确的。使用数字并不
阅读更多2025-01-16
蓝桥杯2020年国赛C/C++C组第7题重复字符串(思维与贪心)
【代码】蓝桥杯2020年国赛C/C++C组第7题重复字符串(思维与贪心)
阅读更多2025-01-16
基于YOLOv8的卫星图像中船只检测系统
（该系统可以根据数据训练出的yolov8的权重文件，运用在其他检测系统上，如火焰检测，口罩检测等等，可以根据检测目标更改UI界面。训练集 3617张验证集 452张测试集 453张。通
阅读更多2025-01-16
人工智能下的MASS服务架构
MaaS（Model as a Service，模型即服务）是一种新型的人工智能服务模式，通过将复杂的AI模型封装为标准化服务，降低了模型的开发和部署门槛，帮助企业快速实现业务场景的智能化升级。
阅读更多2025-01-16
[操作系统] 深入理解约翰·冯·诺伊曼体系
冯诺依曼体系的核心思想是将存储器作为中央组件，所有数据的输入、输出、处理都围绕存储器进行。本质就是将数据进行拷贝，从一个节点拷贝到另一个节点然后进行处理，而拷贝的输入输出设备之间的拷贝效率会影响这个计
阅读更多2025-01-16