快速开发Flink应用

选择“系统 > 权限 > 用户 > 添加用户”，在新增用户界面创建一个机机用户，例如flinkuser。

“用户组”需加入“supergroup”用户组，并关联“System_administrator”角色。

例如，客户端配置文件压缩包为“FusionInsight_Cluster_1_Services_Client.tar”，解压后得到“FusionInsight_Cluster_1_Services_ClientConfig_ConfigFiles.tar”，继续解压该文件。

进入客户端配置文件解压路径“FusionInsight_Cluster_1_Services_ClientConfig_ConfigFiles\Flink\config”，获取相关配置文件。

下载样例工程的Maven工程源码和配置文件，并在本地配置好相关开发工具，可参考通过开源镜像站获取样例工程。

根据集群版本选择对应的分支，下载并获取MRS相关样例工程。

例如本章节场景对应示例为“FlinkStreamJavaExample”样例，获取地址：https://github.com/huaweicloud/huaweicloud-mrs-example/tree/mrs-3.2.0.1/src/flink-examples/flink-examples-security/FlinkStreamJavaExample。

图1 Flink样例工程示例

本地配置好Maven及SDK相关参数后，样例工程会自动加载相关依赖包。

假定用户有某个网站周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Flink的DataStream应用程序实现实时统计总计网购时间超过2个小时的女性网民信息。

开发思路为：

1. 读取文本数据，生成相应DataStream，解析数据生成UserRecord信息。
2. 筛选女性网民上网时间数据信息。
3. 按照姓名、性别进行keyby操作，并汇总在一个时间窗口内每个女性上网时间。
4. 筛选连续上网时间超过阈值的用户，并获取结果。

public class FlinkStreamJavaExample {
    public static void main(String[] args) throws Exception {
        // 打印出执行flink run的参考命令
        System.out.println("use command as: ");
        System.out.println("./bin/flink run --class com.huawei.bigdata.flink.examples.FlinkStreamJavaExample /opt/test.jar --filePath /opt/log1.txt,/opt/log2.txt --windowTime 2");
        System.out.println("******************************************************************************************");
        System.out.println("<filePath> is for text file to read data, use comma to separate");
        System.out.println("<windowTime> is the width of the window, time as minutes");
        System.out.println("******************************************************************************************");

        // 读取文本路径信息，并使用逗号分隔，如果源文件在HDFS中，可配置为详细HDFS路径，例如“hdfs://hacluster/tmp/log1.txt,hdfs://hacluster/tmp/log2.txt”
        final String[] filePaths = ParameterTool.fromArgs(args).get("filePath", "/opt/log1.txt,/opt/log2.txt").split(",");
        assert filePaths.length > 0;

        // windowTime设置窗口时间大小，默认2分钟一个窗口足够读取文本内的所有数据了
        final int windowTime = ParameterTool.fromArgs(args).getInt("windowTime", 2);

        // 构造执行环境，使用eventTime处理窗口数据
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        // 读取文本数据流
        DataStream<String> unionStream = env.readTextFile(filePaths[0]);
        if (filePaths.length > 1) {
            for (int i = 1; i < filePaths.length; i++) {
                unionStream = unionStream.union(env.readTextFile(filePaths[i]));
            }
        }

        // 数据转换，构造整个数据处理的逻辑，计算并得出结果打印出来
        unionStream.map(new MapFunction<String, UserRecord>() {
            @Override
            public UserRecord map(String value) throws Exception {
                return getRecord(value);
            }
        }).assignTimestampsAndWatermarks(
                new Record2TimestampExtractor()
        ).filter(new FilterFunction<UserRecord>() {
            @Override
            public boolean filter(UserRecord value) throws Exception {
                return value.sexy.equals("female");
            }
        }).keyBy(
            new UserRecordSelector()
        ).window(
            TumblingEventTimeWindows.of(Time.minutes(windowTime))
        ).reduce(new ReduceFunction<UserRecord>() {
            @Override
            public UserRecord reduce(UserRecord value1, UserRecord value2)
                    throws Exception {
                value1.shoppingTime += value2.shoppingTime;
                return value1;
            }
        }).filter(new FilterFunction<UserRecord>() {
            @Override
            public boolean filter(UserRecord value) throws Exception {
                return value.shoppingTime > 120;
            }
        }).print();

        // 调用execute触发执行
        env.execute("FemaleInfoCollectionPrint java");
    }

    // 构造keyBy的关键字作为分组依据
    private static class UserRecordSelector implements KeySelector<UserRecord, Tuple2<String, String>> {
        @Override
        public Tuple2<String, String> getKey(UserRecord value) throws Exception {
            return Tuple2.of(value.name, value.sexy);
        }
    }

    // 解析文本行数据，构造UserRecord数据结构
    private static UserRecord getRecord(String line) {
        String[] elems = line.split(",");
        assert elems.length == 3;
        return new UserRecord(elems[0], elems[1], Integer.parseInt(elems[2]));
    }

    // UserRecord数据结构的定义，并重写了toString打印方法
    public static class UserRecord {
        private String name;
        private String sexy;
        private int shoppingTime;

        public UserRecord(String n, String s, int t) {
            name = n;
            sexy = s;
            shoppingTime = t;
        }

        public String toString() {
            return "name: " + name + "  sexy: " + sexy + "  shoppingTime: " + shoppingTime;
        }
    }

    // 构造继承AssignerWithPunctuatedWatermarks的类，用于设置eventTime以及waterMark
    private static class Record2TimestampExtractor implements AssignerWithPunctuatedWatermarks<UserRecord> {

        // add tag in the data of datastream elements
        @Override
        public long extractTimestamp(UserRecord element, long previousTimestamp) {
            return System.currentTimeMillis();
        }

        // give the watermark to trigger the window to execute, and use the value to check if the window elements is ready
        @Override
        public Watermark checkAndGetNextWatermark(UserRecord element, long extractedTimestamp) {
            return new Watermark(extractedTimestamp - 1);
        }
    }
}

1. 在IDEA主页面，选择“File > Project Structures...”进入“Project Structure”页面。
2. 在“Project Structure”页面，选择“Artifacts”，单击“+”并选择“JAR > Empty”。
   图2 添加Artifacts
   
3. 根据实际情况设置Jar包的名称、类型以及输出路径，例如配置为“flink-demo”。
   图3 设置基本信息
   
4. 选中“'FlinkStreamJavaExample' compile output”，右键选择“Put into Output Root”。然后单击“Apply”。
   图4 Put into Output Root
   
5. 最后单击“OK”完成配置。

1. 在IDEA主页面，选择“Build > Build Artifacts...”。
2. 在弹出的菜单中，选择“FlinkStreamJavaExample > Build”开始生成Jar包。
   图5 Build
   
3. Jar包编译成功后，可以从1.c中配置的路径下获取到“flink-demo.jar”文件。

1. 安装MRS集群客户端，例如客户端安装目录为“/opt/hadoopclient”。
2. 将准备集群配置文件中下载的认证凭据压缩包解压缩，并将得到的文件复制到客户端节点中，例如客户端节点的“/opt/hadoopclient/Flink/flink/conf”目录下。
3. 执行以下命令编辑Flink客户端配置参数并保存。

   vi /opt/hadoopclient/Flink/flink/conf/flink-conf.yaml

   将客户端节点的业务IP地址以及Manager的浮动IP地址追加到“jobmanager.web.allow-access-address”配置项中，同时在对应配置添加keytab路径以及用户名。

   ...
   jobmanager.web.allow-access-address: 192.168.64.122,192.168.64.216,192.168.64.234
   ...
   security.kerberos.login.keytab: /opt/hadoopclient/Flink/flink/conf/user.keytab
   security.kerberos.login.principal: flinkuser
   ...

4. 配置安全认证。
   1. 执行以下命令，生成Flink客户端安全认证文件。

      cd /opt/hadoopclient/Flink/flink/bin

      sh generate_keystore.sh

      执行脚本后，需输入一个用于认证的自定义密码。

   2. 配置客户端访问“flink.keystore”和“flink.truststore”文件的路径配置。

      cd /opt/hadoopclient/Flink/flink/conf/

      mkdir ssl

      mv flink.keystore ssl/

      mv flink.truststore ssl/

      vi /opt/hadoopclient/Flink/flink/conf/flink-conf.yaml

      修改如下两个参数的路径详细为相对地址路径。

      security.ssl.keystore: ssl/flink.keystore
      security.ssl.truststore: ssl/flink.truststore

在jar包所在目录下创建“conf”目录，将准备集群配置文件中获取的集群客户端配置文件软件包内“Flink/config”内的配置文件上传至“conf”目录。

在本示例中，源数据文件“log1.txt”、“log2.txt”放置在本地，因此需提前上传至所有Yarn NodeManager实例的节点上的“/opt”目录下，且文件权限配置为755。

cd /opt/hadoopclient/Flink/flink

bin/yarn-session.sh -jm 1024 -tm 1024 -t conf/ssl/

...
Cluster started: Yarn cluster with application id application_1683438782910_0009
JobManager Web Interface: http://192.168.64.10:32261

source /opt/hadoopclient/bigdata_env

cd /opt/hadoopclient/Flink/flink

bin/flink run --class com.huawei.bigdata.flink.examples.FlinkStreamJavaExample /opt/hadoopclient/flink-demo.jar --filePath /opt/log1.txt,/opt/log2.txt --windowTime 2

...
2023-05-26 19:56:52,068 | INFO  | [main] | Found Web Interface host-192-168-64-10:32261 of application 'application_1683438782910_0009'. | org.apache.flink.yarn.YarnClusterDescriptor.setClusterEntrypointInfoToConfig(YarnClusterDescriptor.java:1854)
Job has been submitted with JobID 7647255752b09456d5a580e33a8529f5
Program execution finished
Job with JobID 7647255752b09456d5a580e33a8529f5 has finished.
Job Runtime: 36652 ms

使用flinkuser用户登录FusionInsight Manager，选择“集群 > 服务 > Yarn”，进入Yarn ResourceManager WebUI页面，在“Applications”页面单击作业名称，进入到作业详情页面。

图6 查看Yarn作业详情

对于在session中提交的作业，可以单击“Tracking URL”链接登录Flink原生页面查看作业信息。

图7 查看Flink作业详情

在本示例程序中，单击“Task Managers”，在作业的“Stdout”页签中可查看程序运行结果。

图8 查看程序运行结果