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 后，按以下步骤进行：
tar xzf flink-*.tgz
cd flink-*
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 文件中，可以按需配置，较为简单。就不多赘述了。
注意，这里没有使用 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 集群下运行之后，结果如下所示：
![]()
![]()
![]()
![]()
通过 WebUI 监控界面观察，任务在 Flink 集群中运行所花费的时间在 2s 以内。其运行速度是比较具有诱惑力的。
4.总结
总体来说，Flink 集群的部署较为简单，其 SQL 的 API 编写需要对官方的文档比较熟悉，需要注意的是，在本地运行 Flink 代码，若是要读取远程 HDFS 文件，那么获取 Flink 对象操作环境，需要采用远程接口（HOST & PORT），或者在本地部署一个开发集群环境，将远程数据源提交到本地 Flink 集群环境运行。若是，读取本地文件，则不需要。其中的原因是当你以集群的方式运行，Flink 会检查本地是否有 Flink 集群环境存在，如若不存在，则会出现远程数据源（如：HDFS 路径地址无法解析等错误）。
5.结束语
这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！
