Flink 案例整合

1.概述

　　Flink 1.1.0 版本已经在官方发布了，官方博客于 2016-08-08 更新了 Flink 1.1.0 的变动。在这 Flink 版本的发布，添加了 SQL 语法这一特性。这对于业务场景复杂，依赖于 SQL 来分析统计数据，算得上是一个不错的福利。加上之前有同学和朋友邮件中提到，Flink 官方给的示例运行有困难，能否整合一下 Flink 的案例。笔者通过本篇博客来解答一下相关疑问。

2.内容

2.1 集群部署

　　首先，集群的部署需要 JDK 环境。下载 JDK 以及配置 JAVA_HOME 环境，这里就不详述了，比较简单。然后，我们去下载 Flink 1.1.0 的安装包，进入到下载页面，如下图所示：

　　这里需要注意的是，Flink 集群的部署，本身不依赖 Hadoop 集群，如果用到 HDFS 或是 HBase 中的存储数据，就需要选择对应的 Hadoop 版本。大家可以根据 Hadoop 集群的版本，选择相应的 Flink 版本下载。

　　下载好 Flink 1.1.0 后，按以下步骤进行：

解压 Flink 安装包到 Master 节点

tar xzf flink-*.tgz
cd flink-*

配置 Master 和 Slaves

vi $FLINK_HOME/conf/master
vi $FLINK_HOME/conf/slaves

分发

scp -r flink-1.1.0 hadoop@dn2:/opt/soft/flink
scp -r flink-1.1.0 hadoop@dn3:/opt/soft/flink

　　这里只用了2个 slave 节点。另外，在 flink-conf.yaml 文件中，可以按需配置，较为简单。就不多赘述了。

启动集群

bin/start-cluster.sh

　　注意，这里没有使用 YARN 来启动集群，若是需要使用 YARN 启动集群，可以参考官方文档进行启动。地址

　　Flink 集群启动后，系统有一个 WebUI 监控界面，如下图所示：

2.2 案例

　　这里，我们使用 Flink SQL 的 API 来运行一个场景，对一个销售表做一个聚合计算。这里，笔者将实现代码进行了分解，首先是获取操作 Flink 系统的对象，如下所示：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);

　　接着是读取数据源，并注册为表，如下所示：

CsvTableSource csvTableSource = new CsvTableSource(inPath, new String[] { "trans_id", "part_dt", "lstg_format_name", "leaf_categ_id", "lstg_site_id", "slr_segment_cd", "price", "item_count", "seller_id" },
                    new TypeInformation<?>[] { Types.LONG(), Types.STRING(), Types.STRING(), Types.LONG(), Types.INT(), Types.INT(), Types.FLOAT(), Types.LONG(), Types.LONG() });
tableEnv.registerTableSource("user", csvTableSource);
Table tab = tableEnv.scan("user");

　　这里 inPath 使用了 HDFS 上的数据路径。类型可以在 Hive 中使用 desc 命令查看该表的类型。然后，将“表”转化为数据集，如下所示：

DataSet<KylinSalesDomain> ds = tableEnv.toDataSet(tab, KylinSalesDomain.class);

tableEnv.registerDataSet("user2", ds, "trans_id,part_dt,lstg_format_name,leaf_categ_id,lstg_site_id,slr_segment_cd,price,item_count,seller_id");

Table result = tableEnv.sql("SELECT lstg_format_name as username,SUM(FLOOR(price)) as total FROM user2 group by lstg_format_name");

　　最后，对结果进行存储，这里笔者将结果存在了 HDFS 上。如下所示：

TableSink<?> sink = new CsvTableSink(outPath, "|");
            
result.writeToSink(sink);

env.setParallelism(1);
env.execute("Flink Sales SUM");

　　注意，这里并发数是可以设置的，通过 setParallelism 方法来设置并发数。

　　完整示例，如下所示：

try {
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);

            CsvTableSource csvTableSource = new CsvTableSource(args[0], new String[] { "trans_id", "part_dt", "lstg_format_name", "leaf_categ_id", "lstg_site_id", "slr_segment_cd", "price", "item_count", "seller_id" },
                    new TypeInformation<?>[] { Types.LONG(), Types.STRING(), Types.STRING(), Types.LONG(), Types.INT(), Types.INT(), Types.FLOAT(), Types.LONG(), Types.LONG() });
            tableEnv.registerTableSource("user", csvTableSource);
            Table tab = tableEnv.scan("user");

            DataSet<KylinSalesDomain> ds = tableEnv.toDataSet(tab, KylinSalesDomain.class);

            tableEnv.registerDataSet("user2", ds, "trans_id,part_dt,lstg_format_name,leaf_categ_id,lstg_site_id,slr_segment_cd,price,item_count,seller_id");

            Table result = tableEnv.sql("SELECT lstg_format_name as username,SUM(FLOOR(price)) as total FROM user2 group by lstg_format_name");

            TableSink<?> sink = new CsvTableSink(args[1], "|");
            // write the result Table to the TableSink
            result.writeToSink(sink);

            // execute the program
            env.setParallelism(1);
            env.execute("Flink Sales SUM");
        } catch (Exception e) {
            e.printStackTrace();
        }

　　最后，我们将应用提交到 Flink 集群。如下所示：

flink run flink_sales_sum.jar hdfs://master:8020/user/hive/warehouse/kylin_sales/DEFAULT.KYLIN_SALES.csv hdfs://master:8020/tmp/result3

3.Hive 对比

　　同样的语句，在 Hive 下运行之后，与在 Flink 集群下运行之后，结果如下所示：

Hive 运行结果：

Flink 运行结果：

　　通过 WebUI 监控界面观察，任务在 Flink 集群中运行所花费的时间在 2s 以内。其运行速度是比较具有诱惑力的。

4.总结

　　总体来说，Flink 集群的部署较为简单，其 SQL 的 API 编写需要对官方的文档比较熟悉，需要注意的是，在本地运行 Flink 代码，若是要读取远程 HDFS 文件，那么获取 Flink 对象操作环境，需要采用远程接口（HOST & PORT），或者在本地部署一个开发集群环境，将远程数据源提交到本地 Flink 集群环境运行。若是，读取本地文件，则不需要。其中的原因是当你以集群的方式运行，Flink 会检查本地是否有 Flink 集群环境存在，如若不存在，则会出现远程数据源（如：HDFS 路径地址无法解析等错误）。

5.结束语

　　这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！