Apache-Flink 案例整合

妖狐艹你老母 2021-06-11 15:14 400阅读 0赞

[https://www.cnblogs.com/smartloli/p/5757498.html][https_www.cnblogs.com_smartloli_p_5757498.html]

# 1.概述 #

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

# 2.内容 #

## 2.1 集群部署 ##

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

![666745-20160810151348887-1694910107.png][]

这里需要注意的是，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 启动集群，可以参考官方文档进行启动。[地址][Link 2]

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

![666745-20160810152905699-214512094.png][]

## 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 方法来设置并发数。

完整示例，如下所示：

![复制代码][copycode.gif]

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();
            }

![复制代码][copycode.gif]

最后，我们将应用提交到 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 运行结果：

![666745-20160810164730652-1147348040.png][]

*  Flink 运行结果：

![666745-20160810164829059-588111879.png][]

![666745-20160810165046418-210535868.png][]

![666745-20160810165205981-1507608925.png][]

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

# 4.总结 #

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

# 5.结束语 #

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

[https_www.cnblogs.com_smartloli_p_5757498.html]: https://www.cnblogs.com/smartloli/p/5757498.html
[Link 1]: http://flink.apache.org/news/2016/08/08/release-1.1.0.html
[666745-20160810151348887-1694910107.png]: /images/20210527/880b29d0b35f440ab9405bef9c90a3b1.png
[Link 2]: https://ci.apache.org/projects/flink/flink-docs-master/setup/yarn_setup.html
[666745-20160810152905699-214512094.png]: /images/20210527/fedfa9e95ec24f40a61fed4cdb61a8bb.png
[copycode.gif]: /images/20210527/fdbda052c41845de95014aee7b3d6008.png
[666745-20160810164730652-1147348040.png]: /images/20210527/9c9fd97d4fe94b569c72e1c6db17790d.png
[666745-20160810164829059-588111879.png]: /images/20210527/df27e64ada7042c3bbbf7a2d7113273c.png
[666745-20160810165046418-210535868.png]: /images/20210527/f9abc3056b6b45e3abaeaf00e862fcc9.png
[666745-20160810165205981-1507608925.png]: /images/20210527/0ac398a259504b8fa9eb0089cfc66c94.png