Flink DataStream创建执行环境的正确方式与细节问题

这篇具有很好参考价值的文章主要介绍了Flink DataStream创建执行环境的正确方式与细节问题。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

package com.flink.DataStream.env;

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.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class flinkEnvDemo {
    public static void main(String[] args) throws Exception {
        //TODO 创建一个Flink的配置对象
        Configuration configuration = new Configuration();
        //默认是8081,我们改为8082
        configuration.set(RestOptions.BIND_PORT, "8082");
        //TODO 创建Flink的执行环境
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment
                //.createLocalEnvironment()   //创建本地环境
                //.createRemoteEnvironment()  //远程环境
                //开发过程中直接使用,他会自动判断是本地集群还是远程环境
                //.getExecutionEnvironment();
                //.getExecutionEnvironment(configuration);
                .createLocalEnvironmentWithWebUI(configuration);  //不启动Flink集群也可以有Web UI
        //TODO 流批一体:代码api是同一套 可以指定为流(默认),也可以指定为批
        //TODO 一般不在代码中写死,提交时,指定参数 —Dexeution.runtime-mode=STREAMING/BATCH
        streamExecutionEnvironment.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //TODO 创建FLink的source为socket数据源
        DataStreamSource<String> dataStreamSource = streamExecutionEnvironment.socketTextStream("localhost", 8888);
        //TODO 扁平化+转换+分组+聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> singleOutputStreamOperator = dataStreamSource.flatMap(
                        //使用Lamada表达式实现flatMap接口,当然也可以直接new一个匿名类实现,或者在外部单独定义一个接口实现
                        //泛型第一个是输入类型,第二个是输出类型
                        (String s, Collector<Tuple2<String, Integer>> collector) -> {
                            String[] splitResult = s.split(" ");
                            //循环遍历,将数据转为Tuple类型.spark的rdd算子map: _.map((_,1))
                            for (String word : splitResult) {
                                Tuple2<String, Integer> wordsAndOne = Tuple2.of(word, 1);
                                //使用采集器向下游发送数据
                                collector.collect(wordsAndOne);
                            }
                        })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(
                        (Tuple2<String, Integer> value) -> {
                            return value.f0;
                        }
                ).sum(1);
        //TODO Sink输出
        singleOutputStreamOperator.print();
        //TODO 执行Flink程序,需要抛异常
        streamExecutionEnvironment.execute("Flink Environment Demo");

        //TODO ......
        /**
         * 默认env.execute() 触发一个Flink Job
         * 一个main方法理论上可以指定多个execute,但是没有什么意义,因为指定到第一个就会阻塞掉
         * 但是Flink 提供了异步执行的方式,一个main方法里面executeAsync()的个数 = 生成的Flink Job数
         * */
        //streamExecutionEnvironment.executeAsync();//异步执行
    }
}

文章来源地址https://www.toymoban.com/news/detail-717286.html

到了这里,关于Flink DataStream创建执行环境的正确方式与细节问题的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spring Boot+Flink CDC —— MySQL 同步 Elasticsearch (DataStream方式)

    对于自建 MySQL , 需要先开启 Binlog 写入功能,配置 binlog-format 为 ROW 模式,my.cnf 中配置如下 授权链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant 基于jdk1.8 + springboot2.7.x + elasticsearch7.x 到此就大功告成啦!代码地址:https://gitee.com/qianxkun/lakudouzi-components/tree/

    2024年02月16日
    浏览(39)
  • flink cdc DataStream api 时区问题

    以postgrsql 作为数据源时,Date和timesatmp等类型cdc同步读出来时,会发现一下几个问题: 源表: sink 表: 解决方案:在自定义序列化时进行处理。 java code scala code mysql cdc也会出现上述时区问题,Debezium默认将MySQL中datetime类型转成UTC的时间戳({@link io.debezium.time.Timestamp}),时区是

    2024年02月07日
    浏览(37)
  • 3、flink重要概念(api分层、角色、执行流程、执行图和编程模型)及dataset、datastream详细示例入门和提交任务至on yarn运行

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月12日
    浏览(44)
  • MyBatis SqlSession事务与批量执行正确方式(默认不生效)

    某些情况下会使用MyBatis的SqlSessionFactory.openSession()方法获取SqlSession对象,再进行数据库操作,但默认情况下SqlSession的事务与批量执行均不生效,假如希望使用SqlSession时事务或批量执行能够生效,则需要进行额外的处理 调用org.apache.ibatis.session.SqlSessionFactory接口的以下openSess

    2024年02月09日
    浏览(44)
  • Flink DataStream API详解

    参考:https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/datastream_api.html Data Sources Source是程序读取其输入的位置,您可以使用 env.addSource(sourceFunction) 将Source附加到程序中。Flink内置了许多预先实现的SourceFunction,但是您始终可以通过实现SourceFunction(non-parallel sources)来编写自定

    2024年02月14日
    浏览(40)
  • flink的datastream基本转换

     

    2023年04月13日
    浏览(38)
  • Flink学习——DataStream API

            一个flink程序,其实就是对DataStream的各种转换。具体可以分成以下几个部分: 获取执行环境(Execution Environment) 读取数据源(Source) 定义基于数据的转换操作(Transformations) 定义计算结果的输出位置(Sink) 触发程序执行(Execute)         flink 程序可以在各种上

    2024年02月05日
    浏览(42)
  • 【Apache Flink】Flink DataStream API的基本使用

    Flink DataStream API的基本使用 Flink DataStream API主要用于处理无界和有界数据流 。 无界数据流 是一个持续生成数据的数据源,它没有明确的结束点,例如实时的交易数据或传感器数据。这种类型的数据流需要使用Apache Flink的实时处理功能来连续地处理和分析。 有界数据流 是一个

    2024年02月06日
    浏览(37)
  • Flink|《Flink 官方文档 - DataStream API - 概览》学习笔记

    学习文档:Flink 官方文档 - DataStream API - 概览 学习笔记如下: Flink 的 DataStream API: 数据里的起始是各种 source,例如消息队列、socket 流、文件等; 对数据流进行转换,例如过滤、更新状态、定义窗口、聚合等; 结果通过 sink 返回,例如可以将数据写入文件或标准输出。 Da

    2024年01月23日
    浏览(53)
  • Flink基础之DataStream API

    union联合:被unioin的流中的数据类型必须一致 connect连接:合并的两条流的数据类型可以不一致 connec后,得到的是ConnectedStreams 合并后需要根据数据流是否经过keyby分区 coConnect: 将两条数据流合并为同一数据类型 keyedConnect 目前所使用的大多数Sink, 都是基于2PC的方式来保证状态

    2024年02月05日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包