Flink 案例整合

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介:

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.结束语

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

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
4月前
|
分布式计算 Java API
Flink教程(04)- Flink入门案例
Flink教程(04)- Flink入门案例
53 0
|
3月前
|
数据可视化 JavaScript 关系型数据库
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(五)FineBI可视化
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(五)FineBI可视化
34 0
|
3月前
|
SQL 消息中间件 关系型数据库
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(四)实时计算需求及技术方案
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(四)实时计算需求及技术方案
46 0
|
3月前
|
SQL 消息中间件 分布式数据库
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(三)离线分析
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(三)离线分析
39 0
|
3月前
|
消息中间件 存储 数据采集
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(二)数据源
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(二)数据源
45 0
|
3月前
|
存储 消息中间件 分布式数据库
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(一)案例需求
基于Flume+Kafka+Hbase+Flink+FineBI的实时综合案例(一)案例需求
42 0
|
3月前
|
SQL 分布式计算 Java
2021年最新最全Flink系列教程__Flink综合案例(九)
2021年最新最全Flink系列教程__Flink综合案例(九)
24 0
|
3月前
|
消息中间件 大数据 Kafka
Flink史上最简单双十一实时分析案例
Flink史上最简单双十一实时分析案例
26 0
|
3月前
|
Apache 流计算
电子好书发您分享《Apache Flink 案例集(2022版)》
电子好书发您分享《Apache Flink 案例集(2022版)》
33 1
|
3月前
|
分布式计算 安全 Java
深入理解Java GSS(含kerberos认证及在hadoop、flink案例场景举例)
深入理解Java GSS(含kerberos认证及在hadoop、flink案例场景举例)
41 0