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 案例整合详解大数据

  这里需要注意的是,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 [email protected]:/opt/soft/flink 
scp -r flink-1.1.0 [email protected]:/opt/soft/flink

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

  • 启动集群
bin/start-cluster.sh

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

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

Flink 案例整合详解大数据

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 案例整合详解大数据

  • Flink 运行结果:

Flink 案例整合详解大数据

 Flink 案例整合详解大数据

Flink 案例整合详解大数据

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

4.总结

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

5.结束语

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

原创文章,作者:ItWorker,如若转载,请注明出处:https://blog.ytso.com/9897.html

(0)
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论