09、Flink深入：Flink的各种创建执行环境的方法-6ren

09、Flink深入：Flink的各种创建执行环境的方法

转载作者：大佬之路更新时间：2024-01-07 13:34:24

1. 创建DataSet的执行环境以及WordCount程序

package com.ddkk.hello;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Author ddkk.com  弟弟快看，程序员编程资料站
 * Desc
 * 需求:使用Flink完成WordCount-DataSet
 * 编码步骤
 * 1.准备环境-env
 * 2.准备数据-source
 * 3.处理数据-transformation
 * 4.输出结果-sink
 * 5.触发执行-execute//如果有print,DataSet不需要调用execute,DataStream需要调用execute
 */
public class WordCount1 {
    public static void main(String[] args) throws Exception {
        //老版本的批处理API如下,但已经不推荐使用了
        //1.准备环境-env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //2.准备数据-source
        DataSet<String> lineDS = env.fromElements("itcast hadoop spark","itcast hadoop spark","itcast hadoop","itcast");
        //3.处理数据-transformation
        //3.1每一行数据按照空格切分成一个个的单词组成一个集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        DataSet<String> wordsDS = lineDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的数据
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//将切割处理的一个个的单词收集起来并返回
                }
            }
        });
        //3.2对集合中的每个单词记为1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        DataSet<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是进来一个个的单词
                return Tuple2.of(value, 1);
            }
        });

        //3.3对数据按照单词(key)进行分组
        //0表示按照tuple中的索引为0的字段,也就是key(单词)进行分组
        UnsortedGrouping<Tuple2<String, Integer>> groupedDS = wordAndOnesDS.groupBy(0);

        //3.4对各个组内的数据按照数量(value)进行聚合就是求sum
        //1表示按照tuple中的索引为1的字段也就是按照数量进行聚合累加!
        DataSet<Tuple2<String, Integer>> aggResult = groupedDS.sum(1);

        //3.5排序
        DataSet<Tuple2<String, Integer>> result = aggResult.sortPartition(1, Order.DESCENDING).setParallelism(1);

        //4.输出结果-sink
        result.print();

        //5.触发执行-execute//如果有print,DataSet不需要调用execute,DataStream需要调用execute
        //env.execute();//'execute()', 'count()', 'collect()', or 'print()'.
    }
}

2. 创建DataStream的执行环境以及WordCount程序

package com.ddkk.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Author ddkk.com  弟弟快看，程序员编程资料站
 * Desc
 * 需求:使用Flink完成WordCount-DataStream
 * 编码步骤
 * 1.准备环境-env
 * 2.准备数据-source
 * 3.处理数据-transformation
 * 4.输出结果-sink
 * 5.触发执行-execute
 */
public class WordCount2 {
    public static void main(String[] args) throws Exception {
        //新版本的流批统一API,既支持流处理也支持批处理
        //1.准备环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.准备数据-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark","itcast hadoop spark","itcast hadoop","itcast");

        //3.处理数据-transformation
        //3.1每一行数据按照空格切分成一个个的单词组成一个集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的数据
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//将切割处理的一个个的单词收集起来并返回
                }
            }
        });
        //3.2对集合中的每个单词记为1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是进来一个个的单词
                return Tuple2.of(value, 1);
            }
        });

        //3.3对数据按照单词(key)进行分组
        //0表示按照tuple中的索引为0的字段,也就是key(单词)进行分组
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);
        //3.4对各个组内的数据按照数量(value)进行聚合就是求sum
        //1表示按照tuple中的索引为1的字段也就是按照数量进行聚合累加!
        DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.输出结果-sink
        result.print();

        //5.触发执行-execute
        env.execute();//DataStream需要调用execute
    }
}

3. Lambda版的WordCount程序

package com.ddkk.hello;

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

import java.util.Arrays;

/**
 * Author ddkk.com  弟弟快看，程序员编程资料站
 * Desc
 * 需求:使用Flink完成WordCount-DataStream--使用lambda表达式
 * 编码步骤
 * 1.准备环境-env
 * 2.准备数据-source
 * 3.处理数据-transformation
 * 4.输出结果-sink
 * 5.触发执行-execute
 */
public class WordCount3_Lambda {
    public static void main(String[] args) throws Exception {
        //1.准备环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.准备数据-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3.处理数据-transformation
        //3.1每一行数据按照空格切分成一个个的单词组成一个集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        //lambda表达式的语法:
        // (参数)->{方法体/函数体}
        //lambda表达式就是一个函数,函数的本质就是对象
        DataStream<String> wordsDS = linesDS.flatMap(
                (String value, Collector<String> out) -> Arrays.stream(value.split(" ")).forEach(out::collect)
        ).returns(Types.STRING);

        //3.2对集合中的每个单词记为1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        /*DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(
                (String value) -> Tuple2.of(value, 1)
        ).returns(Types.TUPLE(Types.STRING, Types.INT));*/
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(
                (String value) -> Tuple2.of(value, 1)
                , TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})
        );

        //3.3对数据按照单词(key)进行分组
        //0表示按照tuple中的索引为0的字段,也就是key(单词)进行分组
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
        //KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy((KeySelector<Tuple2<String, Integer>, String>) t -> t.f0);
        KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);

        //3.4对各个组内的数据按照数量(value)进行聚合就是求sum
        //1表示按照tuple中的索引为1的字段也就是按照数量进行聚合累加!
        DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.输出结果-sink
        result.print();

        //5.触发执行-execute
        env.execute();
    }
}

4. 在windos本地创建含有WebUI的Flink执行环境

4.1. 导入依赖

<!--flink的web包，可以在本地idea执行程序时显示web界面-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-runtime-web_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>

4.2. 具体代码

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.text.SimpleDateFormat;

/**
 * @date: 2021/7/1
 *  @Author ddkk.com  弟弟快看，程序员编程资料站
 * @desc: 在windos本地创建Flink的含有WebUI的执行环境
 */
public class WebUIForIDEA {

    public static void main(String[] args) throws Exception {

        // 使用流的执行环境类创建本地并含有WebUI的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env
                .addSource(new SourceFunction<String>() {

                    boolean flag = true;
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                    @Override
                    public void run(SourceContext<String> sourceContext) throws Exception {
                        while (flag) {
                            sourceContext.collect(sdf.format(System.currentTimeMillis()));
                            Thread.sleep(1000);
                        }
                    }

                    @Override
                    public void cancel() {
                        flag = false;
                    }
                })
                .print("WebUIForIDEA >>>>>");

        // 启动执行
        env.execute("WebUIForIDEA");
    }
}

4.3. 执行效果

访问url：http://localhost:8081/#/job-manager/stdout

5. 在Yarn上运行Flink程序

步骤一：编写代码 ：

package com.ddkk.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Author ddkk.com  弟弟快看，程序员编程资料站
 * Desc
 * 需求:使用Flink完成WordCount-DataStream--使用lambda表达式--修改代码使适合在Yarn上运行
 * 编码步骤
 * 1.准备环境-env
 * 2.准备数据-source
 * 3.处理数据-transformation
 * 4.输出结果-sink
 * 5.触发执行-execute//批处理不需要调用!流处理需要
 */
public class WordCount4_Yarn {
    public static void main(String[] args) throws Exception {
        //获取参数
        ParameterTool params = ParameterTool.fromArgs(args);
        String output = null;
        if (params.has("output")) {
            output = params.get("output");
        } else {
            output = "hdfs://node1:8020/wordcount/output_" + System.currentTimeMillis();
        }

        //1.准备环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.准备数据-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3.处理数据-transformation
        DataStream<Tuple2<String, Integer>> result = linesDS
                .flatMap(
                        (String value, Collector<String> out) -> Arrays.stream(value.split(" ")).forEach(out::collect)
                ).returns(Types.STRING)
                .map(
                        (String value) -> Tuple2.of(value, 1)
                ).returns(Types.TUPLE(Types.STRING, Types.INT))
                //.keyBy(0);
                .keyBy((KeySelector<Tuple2<String, Integer>, String>) t -> t.f0)
                .sum(1);

        //4.输出结果-sink
        result.print();

        //如果执行报hdfs权限相关错误,可以执行 hadoop fs -chmod -R 777  /
        System.setProperty("HADOOP_USER_NAME", "root");//设置用户名
        //result.writeAsText("hdfs://node1:8020/wordcount/output_"+System.currentTimeMillis()).setParallelism(1);
        result.writeAsText(output).setParallelism(1);

        //5.触发执行-execute
        env.execute();
    }
}

步骤二：打包上传

步骤三：提交执行

Apache Flink 1.12 Documentation: Execution Mode (Batch/Streaming)

执行命令：

/export/server/flink/bin/flink run -Dexecution.runtime-mode=BATCH -m yarn-cluster -yjm 1024 -ytm 1024 -c com.ddkk.hello.WordCount4_Yarn /root/wc.jar --output hdfs://node1:8020/wordcount/output_xx

步骤四：在web页面观察提交的程序

http://node1:8088/cluster

http://node1:50070/explorer.html#/

或者在Standalone模式下使用web界面提交

注意事项 ：

写入HDFS如果存在权限问题进行如下设置：

hadoop fs -chmod -R 777 /

并在代码中添加:

System.setProperty("HADOOP_USER_NAME", "root")

文章推荐： 10、Flink 笔记 - window窗口（一）

文章推荐： 01、Flink 笔记 - 概述和入门案例

文章推荐： 08、Flink深入：Flink的API说明和pom文件汇总

文章推荐： 63、Flink深入：Flink中通用RedisUtil工具类

macos - 执行 wine != 执行 `which wine`
我有一个“有趣”的问题，即以两种不同的方式运行 wine 会导致: $> wine --version /Applications/Wine.app/Contents/Resources/bin/wi
javascript - CRONTAB 执行 Python，使用 puppeteer 执行 Node 来进行网页抓取不起作用
我制作了这个网络抓取工具来获取网页中的表格。我使用 puppeteer (不知道 crontab 有问题)、Python 进行清理并处理数据库的输出但令我惊讶的是，当我执行它时 */50 * * *
javascript - 对 javascript 函数的 Objective-C 调用何时被调用/执行，何时不被调用/执行？
JavaScript 是否被调用或执行取决于什么？准确地说，我有两个函数，它们都以相同的方式调用: [self.mapView stringByEvaluatingJavaScriptFromStri
python - 为什么使用 statsmodels 执行 OLS 和使用 scikit 执行 PooledOLS 时会得到相同的结果？
我目前正在使用 python 做一个机器学习项目(这里是初学者，从头开始学习一切)。只是想知道 statsmodels 的 OLS 和 scikit 的 PooledOlS 使用我拥有的相同面板数据
c# - 通过 Enterprise Guide 执行 SAS 和从 .Net 执行 IOM 之间的区别
在使用集成对象模型 (IOM) 后，我可以执行 SAS 代码并将 SAS 数据集读入 .Net/C# 数据集 here . 只是好奇，使用 .Net 作为 SAS 服务器的客户端与使用 Enterpr
javascript - jQuery 不会使用 animate : top 200px function. 执行，但它会使用 animate: height 执行
有一些直接的 jQuery 在单击时隐藏打开的 div 未显示，但仍将高度添加到导航中以使其看起来好像要掉下来了。这个脚本工作正常: $(document).ready(funct
java - 为什么我的代码使用 'IF' 执行 'ELSE' 和 '==' ，但不使用 '.equals' 执行？
这个问题已经有答案了: How do I compare strings in Java? (23 个回答) 已关闭 4 年前。这里是 Java 新手，我正在使用 NetBeans 尝试一些简单的代
python - Keras 2.0.8 仅使用 Python 3.x 执行 1 个 epoch，使用 2.x 执行 10 个
如果我将它切换到 Python 2.x，它执行 10。这是为什么？训练逻辑回归模型 import keras.backend as
JavaScript 执行
我有两个脚本，它们包含在 HTML 正文中。在第一个脚本中，我初始化一个 JS 对象，该对象在第二个脚本标记中引用。 ... obj.a = 1000; obj.
执行@number时的Java链接列表错误消息
每当我运行该方法时，我都会收到一个带有数字的错误以下是我的代码。 public String getAccount() { String s = "Listing the accounts";
java - 执行 while 循环以显示菜单
我已经用 do~while(true) 创建了我的菜单；但是每次用户输入一个数字时，它不会运行程序，而是再次显示菜单!你怎么看？ //我的主要方法 public static void main(St
ipython - 执行/命令完成时通知
执行命令后，如何让IPython通知我？我可以使用铃声/警报还是通过弹出窗口获取它？我正在OS X 10.8.5的iTerm上运行Anaconda。最佳答案使用最新版本的iTerm，您可以在she
java - Swing 执行
您好，我刚刚使用菜单栏为 Swing 编写了代码。但是问题出现在运行中。我输入: javac Menu.java java Menu 它没有给出任何错误，但 GUI 没有显示。这是我的源代码以供引用:
.net - 执行.NET应用程序时验证Authenticode签名
我觉得这里缺少明显的东西，但是我看不到它写在任何地方。我使用Authenticode证书对可执行文件进行签名，但是当我开始学习有关它的更多信息时，我对原样的值(value)提出了质疑。签名的exe
按钮单击事件上的 JavaScript 执行
我正在设计一个应用程序，它使用 DataTables 中的预定义库来创建数据表。我想对数据表执行删除操作，为此应在按钮单击事件上执行 java 脚本。 $(document).ready(functi
Haskell - 执行 while 循环
我是 Haskell 新手，如果有人愿意帮助我，我会很高兴!我试图让这个程序与 do while 循环一起工作。第二个 getLine 命令的结果被放入变量 goGlenn 中，如果 goGlenn
java - 执行 while 循环时出现问题
我有一个用 swing 实现迷你游戏的程序，在主类中我有一个循环，用于监听游戏 map 中的 boolean 值。使用 while 实现的循环不会执行一条指令，如果它是唯一的一条指令，我不知道为什么。
java - 执行.jar时将OJBDC添加到类路径
我正在尝试开发一个连接到 Oracle 数据库并执行函数的 Java 应用程序。如果我在 Eclipse 中运行该应用程序，它可以工作，但是当我尝试在 Windows 命令提示符中运行 .jar 时，
java future 执行
我正在阅读有关 Java 中的 Future 和 javascript 中的 Promises 的内容。下面是我作为示例编写的代码。我的问题是分配给 future 的任务什么时候开始执行？当如下行创
java - 执行 && 最有效的方法？
我有一个常见的情况，您有两个变量(xSpeed 和 ySpeed)，当它们低于 minSpeed 时，我想将它们独立设置为零，并在它们都为零时退出。最有效的方法是什么？目前我有两种方法(方法2更干净

大佬之路

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城