您现在的位置是:首页 > 正文

Flink使用(一) Streaming API处理有界流

2024-02-01 03:57:07阅读 3

前提:

目前,Flink版本支持,批流处理使用一套API完成。

即,使用DataStream API既能处理流数据,又能处理批数据(有界流)。

如何使用DataStream API处理数据,并且以BATCH的方式运行呢?

方式一:在程序中指定运行模式(不推荐,调试可以):

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

方式二:

提交任务时,使用bin/flink/run -Dexecution.runtime-mode=BATCH 指定BATCH方式运行程序。

使用BATCH方式运行的目的?

我理解是将最终的结果进行输出,而不是获取一条记录后就输出一条结果。

示范例子

例子数据源word.txt,只有三行数据

hello word
hello flink
hello java

正常情况下,使用DataStream API对数据进行流式处理。

package org.galaxy.foundation.core.courses;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

 
public class Stream10 {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lineDataStreamSource = env.readTextFile("/tmp/foundation-core/words.txt");
       SingleOutputStreamOperator<Tuple2<String,Long>> wordAndOneTuple=  lineDataStreamSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));


          KeyedStream<Tuple2<String, Long>, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

        SingleOutputStreamOperator<Tuple2<String, Long>> sum = wordAndOneKeyedStream.sum(1);
        sum.print();
        env.execute();
    }

}

打印结果:

3> (hello,1)
2> (java,1)
7> (flink,1)
6> (word,1)
3> (hello,2)
3> (hello,3)

可以看出,每来一条记录,即产生一条输出。

重点来了,使用DataStream API,以BATCH方式处理数据

只要在env初始化的下面增加一行

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

重新运行,打印结果:

6> (word,1)
2> (java,1)
7> (flink,1)
3> (hello,3)

可以看出,将最终的汇总结果,一次完整的打印出来

网站文章

  • Python 字典(2)

    Python 字典(2)

    一、遍历字典一个字典可能会包含多个键-值对,字典可以以多种方式存储信息,因此有多种遍历字典的方式,比如键-值对、键、值。  1、遍历所有的键-值对  user_01 = {'username':'tizer','first_name':'joker','last_name':'pon',}  以上面的字典为例,遍历键-值对:  使用函数 items():...

    2024-02-01 03:56:38
  • element-ui的el-table,使用sortablejs改造成可拖拽排序

    element-ui的el-table,使用sortablejs改造成可拖拽排序

    element-ui的el-table,使用sortablejs改造成可拖拽排序。

    2024-02-01 03:56:31
  • 操作系统存储器管理实验报告_别只做CRUD程序员,收下这篇操作系统干货(一)!...

    操作系统存储器管理实验报告_别只做CRUD程序员,收下这篇操作系统干货(一)!...

    一、操作系统概念1.计算机系统组成任何一个计算机系统都是由两部分组成:计算机硬件和计算机软件。- 计算机硬件通常是由中央处理机(运算器和控制器)、存储器、输入设备和输出设备等部件组成。- 计算机软件包括系统软件和应用软件。系统软件如操作系统、多种语言处理程序( 汇编和编译程序等 )、连接装配程序、系统实用程序、多种工具软件等; 应用软件为多种应用目的而编制的程序。2.操作系统的非形...

    2024-02-01 03:56:24
  • 软件测试-测试的时候获取验证码的方式

    软件测试-测试的时候获取验证码的方式

    问开发他们创建的表在哪里,找到那个验证码的表,打开,就可以看到测试的时候用的电话号码返回的验证码。在测试服打开f12,再点击获取验证码,通过network找到相对应的url,找到验证码。我的学习之路任重道远,咱们下期见~~~

    2024-02-01 03:55:50
  • Java基础(方法重写——override)

    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~开发工具与关键技术:IDEA;...

    2024-02-01 03:55:38
  • 每日一题——v-if和v-show的理解

    每日一题——v-if和v-show的理解

    2024-02-01 03:55:09
  • winform前后端框架_Vue + .NetCore前后端分离的快速发开框架

    框架核心快速开发(基础功能全部由代码生成器生成)支持前端、后台自定义业务代码扩展,后台提供了大量常用扩展与通用类前端、后台提供了近300个扩展方法与属性,开发人员可在此功能上编写扩展自定义业务代码代码...

    2024-02-01 03:55:03
  • LeetCode--相关二叉树的例题与理解

    1、相同的树 给定两个二叉树,编写一个函数来检验它们是否相同。 如果两个树在结构上相同,并且节点具有相同的值,则认为它们是相同的。 示例 1: 输入: 1 1 / \ / \ 2 3 2 3 [1,2,3], [1,2,3] 输出: true 示例 2: 输入:...

    2024-02-01 03:54:57
  • C#软件开发实例.私人订制自己的屏幕截图工具(七)添加放大镜的功能

    C#软件开发实例.私人订制自己的屏幕截图工具(七)添加放大镜的功能

    上一篇:C#软件开发实例.私人订制自己的屏幕截图工具(六)添加配置管理功能 由于截图时可能需要精确截取某一部分,所以需要放大镜的功能,这样截取的时候才更容易定位截图的位置。 添加PictureBox,name属性设置为“pictureBox_zoom”; 在“Form1_Load”事件处理函数中添加以下代码: //设置放大镜的大小 this.pictureBox_zoo...

    2024-02-01 03:54:27
  • 快速复现利用Log4j漏洞启动windows计算器

    快速复现利用Log4j漏洞启动windows计算器

    了解关于漏洞的描述,可以参考Vulnerability Affecting Multiple Log4j Versions Permits RCE Exploit根据文章描述,首先下载JDK1.8u1...

    2024-02-01 03:54:20