# 快速开发Flink应用 Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。 Flink最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。 Flink整个系统包含三个部分： - Client Flink Client主要给用户提供向Flink系统提交用户任务（流式作业）的能力。 - TaskManager Flink系统的业务执行节点，执行具体的用户任务。TaskManager可以有多个，各个TaskManager都平等。 - JobManager Flink系统的管理节点，管理所有的TaskManager，并决策用户任务在哪些TaskManager执行。JobManager在HA模式下可以有多个，但只有一个主JobManager。 MRS对外提供了多个Flink组件的应用开发样例工程，本实践用于指导您创建MRS集群后，获取并导入样例工程并在本地进行编译调测，用于实现Flink DataStream程序处理数据。 #### 创建MRS Flink集群 1. 购买一个包含有Hive组件的MRS集群，详情请参见[购买自定义集群](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0513.html)。 ![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 本文以购买的MRS 3.2.0-LTS.1版本的集群为例，组件包含Hadoop、Flink组件，集群开启Kerberos认证。 2. 单击"立即购买"，等待MRS集群创建成功。 #### 准备集群配置文件 1. 集群创建成功后，登录FusionInsight Manager创建用于提交Flink作业的集群用户。 选择"系统 \> 权限 \> 用户 \> 添加用户"，在新增用户界面创建一个机机用户，例如**flinkuser**。 "用户组"需加入"supergroup"用户组，并关联"System_administrator"角色。 2. 选择"系统 \> 权限 \> 用户"，在用户名为"flinkuser"的操作列选择"更多 \> 下载认证凭据"下载认证凭据文件，保存后解压得到该用户的"user.keytab"文件与"krb5.conf"文件。 3. 选择"集群 \> 概览 \> 更多 \> 下载客户端"，"选择客户端类型"设置为"仅配置文件"，单击"确定"，等待客户端文件包生成后根据浏览器提示下载客户端到本地并解压。 例如，客户端配置文件压缩包为"FusionInsight_Cluster_1_Services_Client.tar"，解压后得到"FusionInsight_Cluster_1_Services_ClientConfig_ConfigFiles.tar"，继续解压该文件。 进入客户端配置文件解压路径"FusionInsight_Cluster_1_Services_ClientConfig_ConfigFiles\\Flink\\config"，获取相关配置文件。 #### 获取样例工程 1. 通过开源镜像站获取样例工程。 下载样例工程的Maven工程源码和配置文件，并在本地配置好相关开发工具，可参考[通过开源镜像站获取样例工程](https://support.huaweicloud.com/devg3-mrs/mrs_07_010002.html)。 根据集群版本选择对应的分支，下载并获取MRS相关样例工程。 例如本章节场景对应示例为"FlinkStreamJavaExample"样例，获取地址：。 2. 本地使用IDEA工具导入样例工程，等待Maven工程下载相关依赖包，具体操作可参考[配置并导入样例工程](https://support.huaweicloud.com/devg3-mrs/mrs_07_410007.html)。 图1Flink样例工程示例 ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001608644369.png) 本地配置好Maven及SDK相关参数后，样例工程会自动加载相关依赖包。 3. 在本示例中，将开发的DataStream程序通过Flink客户端提交运行，因此在代码中不需单独进行安全认证。 假定用户有某个网站周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Flink的DataStream应用程序实现实时统计总计网购时间超过2个小时的女性网民信息。 源数据内容如下，日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为","。 - log1.txt：周六网民停留日志。 ``` LiuYang,female,20 YuanJing,male,10 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60 ``` log2.txt：周日网民停留日志。 ``` LiuYang,female,20 YuanJing,male,10 CaiXuyu,female,50 FangBo,female,50 GuoYijun,male,5 CaiXuyu,female,50 Liyuan,male,20 CaiXuyu,female,50 FangBo,female,50 LiuYang,female,20 YuanJing,male,10 FangBo,female,50 GuoYijun,male,50 CaiXuyu,female,50 FangBo,female,60 ``` 开发思路为： 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(" is for text file to read data, use comma to separate"); System.out.println(" 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 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() { @Override public UserRecord map(String value) throws Exception { return getRecord(value); } }).assignTimestampsAndWatermarks( new Record2TimestampExtractor() ).filter(new FilterFunction() { @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() { @Override public UserRecord reduce(UserRecord value1, UserRecord value2) throws Exception { value1.shoppingTime += value2.shoppingTime; return value1; } }).filter(new FilterFunction() { @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> { @Override public Tuple2 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 { // 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. 在IntelliJ IDEA中，配置工程的Artifacts信息。 1. 在IDEA主页面，选择"File \> Project Structures..."进入"Project Structure"页面。 2. 在"Project Structure"页面，选择"Artifacts"，单击"+"并选择"JAR \> Empty"。 图2添加Artifacts ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001608448325.png "点击放大") 3. 根据实际情况设置Jar包的名称、类型以及输出路径，例如配置为"flink-demo"。 图3设置基本信息 ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001558049118.png "点击放大") 4. 选中"'FlinkStreamJavaExample' compile output"，右键选择"Put into Output Root"。然后单击"Apply"。 图4Put into Output Root ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001558208578.png "点击放大") 5. 最后单击"OK"完成配置。 2. 生成Jar包。 1. 在IDEA主页面，选择"Build \> Build Artifacts..."。 2. 在弹出的菜单中，选择"FlinkStreamJavaExample \> Build"开始生成Jar包。 图5Build ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001608288785.png) 3. Jar包编译成功后，可以从[1.c]中配置的路径下获取到"flink-demo.jar"文件。 3. 安装并配置Flink客户端。 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 ``` 4. 将[2]中生成的Jar包上传到Flink客户端节点相关目录下，例如上传至"/opt/hadoopclient"。 在jar包所在目录下创建"conf"目录，将[准备集群配置文件]中获取的集群客户端配置文件软件包内"Flink/config"内的配置文件上传至"conf"目录。 5. 将应用程序待处理的源数据文件上传至NodeManager实例所在节点。 在本示例中，源数据文件"log1.txt"、"log2.txt"放置在本地，因此需提前上传至所有Yarn NodeManager实例的节点上的"/opt"目录下，且文件权限配置为755。 6. 在Flink客户端下通过**yarn session** 命令启动Flink集群。 **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 ``` 7. Flink集群启动成功后，重新打开一个客户端连接窗口，进入Flink客户端目录运行程序。 **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 ``` 8. 查看运行结果。 使用**flinkuser**用户登录FusionInsight Manager，选择"集群 \> 服务 \> Yarn"，进入Yarn ResourceManager WebUI页面，在"Applications"页面单击作业名称，进入到作业详情页面。 图6查看Yarn作业详情 ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001558912226.png "点击放大") 对于在session中提交的作业，可以单击"Tracking URL"链接登录Flink原生页面查看作业信息。 图7查看Flink作业详情 ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001559071998.png "点击放大") 在本示例程序中，单击"Task Managers"，在作业的"Stdout"页签中可查看程序运行结果。 图8查看程序运行结果 ![](https://support.huaweicloud.com/devg-lts-mrs/zh-cn_image_0000001559072006.png "点击放大")