从 0 到 1 通过 Flink + Tablestore 进行大数据处理与分析-阿里云开发者社区

开发者社区> 阿里云实时计算Flink> 正文
登录阅读全文

从 0 到 1 通过 Flink + Tablestore 进行大数据处理与分析

简介: 阿里云实时计算Flink版是一套基于 Apache Flink 构建的⼀站式实时大数据分析平台。在大数据场景下,实时计算 Flink 可提供端到端亚秒级实时数据流批处理能力。表格存储 Tablestore (又名 OTS)是阿里云自研的多模型结构化数据存储,可提供海量结构化数据的存储、查询分析服务。表格存储的双引擎架构支持千万TPS和毫秒级延迟的服务能力,可作为大数据计算的极佳上下游存储。

一、前言

实时计算Flink版是一套基于 Apache Flink 构建的⼀站式实时大数据分析平台。在大数据场景下,实时计算 Flink 可提供端到端亚秒级实时数据流批处理能力。

表格存储 Tablestore (又名 OTS)是阿里云自研的多模型结构化数据存储,可提供海量结构化数据的存储、查询分析服务。表格存储的双引擎架构支持千万TPS和毫秒级延迟的服务能力,可作为大数据计算的极佳上下游存储。

本文章将以商品订单场景为基础,介绍如何从 0 到 1 通过 Flink+Tablestore 进行大数据分析。

二、场景介绍

某大型连锁超市会实时产生大量的消费数据,通过分析这些数据可获取到商品售卖热度、门店经营状态极具商业价值的信息,便于辅助经营者的商业决策。现需要设计一套方案,获取每分钟不同商品类别的 GMV (商品交易总额)。

本文采用云数据库 RDS 作为商品消费订单的存储库,通过 mysql-cdc connector 作为源表接入实时计算Flink。以表格存储 Tablestore 作为商品元信息的存储库,通过 OTS connector 作为维表接入实时计算Flink。配置流计算作业任务计算商品 GMV ,并写入表格存储 Tablestore 结果表进行保存。方案架构图如下:

img

三、实现步骤准备工作

  1. 创建 RDS MySQL 实例。创建步骤请参考创建 RDS MySQL 实例
  2. 开通表格存储服务,并创建按量模式实例。详情请参考开通表格存储服务创建实例
  3. 开通实时计算服务,并购买Flink全托管集群。

1. 数据源准备

1. 登录RDS控制台,登陆数据管理DMS,创建consume_record表。

img

CREATE TABLE `consume_record` (
    `consume_id` varchar(20) NOT NULL,
    `product_id` varchar(20) NOT NULL,
    `consume_time` bigint(20) NOT NULL,
    `consume_name` varchar(20) NOT NULL,
    `consume_phone` varchar(20) NOT NULL,
    PRIMARY KEY (`consume_id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8
COMMENT='消费记录数据源表';

2. 登录 Tablestore 控制台。创建 gmv_result、product 两张表。

说明:Tablestore数据表是schema free的,只需要定义主键,无需定义属性列。关于创建表步骤请参考创建数据表

product表。作为流计算任务的维表,存储商品元数据信息。

字段名字段类型是否主键描述
product_IDSTRING商品ID
priceBIGINT商品单价
product_typeSTRING商品类别

gmv_result表。作为流计算任务的结果表,存储商品交易总额的计算结果。

字段名字段类型是否主键描述
product_typeSTRING商品类型
gmv_timeSTRING统计时间
total_priceBIGINT商品交易总额

img

img

2. Flink 作业配置

1. 登陆 Flink 全托管控制台,创建项目并创建作业,作业名 gmv_pre_aggregation。

img

作业脚本

-- mysql-cdc 源表
CREATE TEMPORARY TABLE consume_record (
  `consume_id` VARCHAR(20),
  `product_id` VARCHAR(20),
  `consume_time` BIGINT,
  `consume_name` VARCHAR(20),
  `consume_phone` VARCHAR(20),
  PRIMARY KEY(consume_id)  NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '',
  'port' = '3306',
  'username' = '',
  'password' = '',
  'database-name' = '',
  'table-name' = ''
);
-- tablestore 维表
CREATE TEMPORARY TABLE product (
  product_ID STRING,
  price BIGINT,
  product_type STRING,
  PRIMARY KEY (product_ID) NOT ENFORCED
) WITH (
  'connector' = 'ots',
  'endPoint' = '',
  'instanceName' = '',
  'tableName' = '',
  'accessId' = '',
  'accessKey' = ''
);
-- tablestore 结果表
CREATE TEMPORARY TABLE gmv_result (
   product_type STRING,
   gmv_time BIGINT,
   total_price BIGINT,
   PRIMARY KEY (product_type) NOT ENFORCED
) WITH (
  'connector' = 'ots',
  'endPoint' = '',
  'instanceName' = '',
  'tableName' = '',
  'accessId' = '',
  'accessKey' = '',
  'valueColumns' = 'gmv_time,total_price'
);
INSERT INTO `gmv_result`
select 
  d.product_type, 
  UNIX_TIMESTAMP(s.consume_time,'yy-MM-dd') as gmv_time,
  sum (d.price) as total_price
from
  `consume_record` as s
  JOIN `product` for system_time as of proctime() as d
  on s.product_id = d.product_ID
GROUP BY d.product_type,UNIX_TIMESTAMP(s.consume_time,'yy-MM-dd')

2. 上线作业。

3. 结果展示

作业 gmv_pre_aggregation 直接写入 gmv 结果到 Tablestore,登录 Tablestore 控制台查询 gmv_result 表 即可获取商品交易总额结果。

img

四、方案改进

采用 Flink + Tablestore 方案很好地实现了商品交易总额的计算。然而 Flink 作业中固定了源表与维表参与计算的字段,如果业务需求发生变化,不得不重新设计作业脚本,再次进行作业下线上线操作,非常繁琐。所以将对上述方案做一下改进:Flink 流计算作业中仅负责关联消费记录和商品信息数据并写入 Tablestore 中。再通过Tablestore 的功能之一多元索引进行数据分析。多元索引基于倒排索引、列式存储、空间索引等,可解决大数据的复杂查询、分析聚合等需求。通过在数据表上建立多元索引,可实现全文检索、前缀查询、模糊查询、组合查询、统计聚合等功能。方案二架构图如下:

img

五、实现步骤

1. 数据源准备

登录 Tablestore 控制台。创建 consume_product 表。

说明:Tablestore数据表是schema free的,只需要定义主键,无需定义属性列。关于创建表步骤请参考创建数据表

consume_product表。作为流计算任务结果表,保存了商品消费信息与商品元数据信息。

字段名字段类型是否主键描述
consume_idSTRING消费ID(主键)
product_IDSTRING商品ID
product_numBIGINT(10)商品数量
priceDOUBLE商品单价
consume_timeBIGINT(20)消费时间
product_typeSTRING商品类别

2. Flink 作业配置

1. 登陆 Flink 全托管控制台,创建项目并创建作业,作业名 gmv_post_aggregation。

-- mysql-cdc 源表,与之前相同
CREATE TEMPORARY TABLE consume_record (
  `consume_id` VARCHAR(20),
  `product_id` VARCHAR(20),
  `consume_time` BIGINT,
  `consume_name` VARCHAR(20),
  `consume_phone` VARCHAR(20),
  PRIMARY KEY(consume_id)  NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '',
  'port' = '3306',
  'username' = '',
  'password' = '',
  'database-name' = '',
  'table-name' = ''
);
-- tablestore 维表,与之前相同
CREATE TEMPORARY TABLE product (
  product_ID STRING,
  price BIGINT,
  product_type STRING,
  PRIMARY KEY (product_ID) NOT ENFORCED
) WITH (
  'connector' = 'ots',
  'endPoint' = '',
  'instanceName' = '',
  'tableName' = '',
  'accessId' = '',
  'accessKey' = ''
);
-- tablestore 结果表
CREATE TEMPORARY TABLE consume_product (
   consume_id STRING,
   product_id STRING,
   price BIGINT,
   consume_time BIGINT,
   consume_name STRING,
   consume_phone STRING,
   PRIMARY KEY (consume_id,product_id) NOT ENFORCED
) WITH (
  'connector' = 'ots',
  'endPoint' = '',
  'instanceName' = '',
  'tableName' = '',
  'accessId' = '',
  'accessKey' = '',
  'valueColumns' = 'price,consume_time,consume_name,consume_phone'
);
insert into consume_product
select s.consume_id,d.product_ID as product_id,d.price,
        UNIX_TIMESTAMP(s.consume_time,'yy-MM-dd') as consume_time,
        s.consume_name,s.consume_phone
        from `consume_record` as s 
        join `product` for system_time as of proctime() as d
        on s.product_id = d.product_ID

2. 上线作业

3. 创建多元索引

登录 Tablestore 控制台,在 consume_product 表上建立多元索引。可通过控制台 SQL 查询或 SDK 分析获取商品交易总额信息。

说明:多元索引创建步骤请参考创建及使用多元索引

创建索引

img

4. 结果展示

SQL查询

img

SDK查询

 SearchRequest searchRequest = SearchRequest.newBuilder()
                .tableName("consume_product")
                .indexName("consume_product_index")
                .searchQuery(SearchQuery.newBuilder()
                        .query(QueryBuilders.matchAll())
                        .addGroupBy(GroupByBuilders.groupByField("groupByProductID","product_id").addSubAggregation(
                                AggregationBuilders.sum("sumagg","price")
                        ))
                        .build())
                .build();
        SearchResponse searchResponse = syncClient.search(searchRequest);
        for(GroupByFieldResultItem item : searchResponse.getGroupByResults().getAsGroupByFieldResult("groupByProductID").getGroupByFieldResultItems()){
            System.out.println("商品ID:"+item.getKey()+",交易总额:"+item.getSubAggregationResults().getAsSumAggregationResult("sumagg").getValue());
        }

SDK查询结果

商品ID:A001,交易总额:20.0
商品ID:A002,交易总额:40.0
商品ID:A004,交易总额:20.0
商品ID:A003,交易总额:5.0
商品ID:A005,交易总额:15.0
商品ID:A006,交易总额:5.0
商品ID:A008,交易总额:5.0

六、总结

本篇文章演示了基于 Flink + Tablestore 方案在大数据计算场景下的应用。后续,我们会推出 Flink on Tablestore 系列文章,并针对维表和结果表场景推出最佳实践文章。

希望本次文章对你有帮助,如果希望继续交流,可以加入表格存储技术交流群,可搜索群号『11789671』或『23307953』。

本文转载自:https://developer.aliyun.com/article/790005


更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~

image.png

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制T恤;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

一套基于Apache Flink构建的一站式、高性能实时大数据处理平台,广泛适用于流式数据处理、离线数据处理、DataLake计算等场景。

官方博客
最新文章
相关文章
链接