基于 MySQL + Tablestore 分层存储架构的大规模订单系统实践-数据流计算篇

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
实时计算 Flink 版,5000CU*H 3个月
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
简介: 背景在订单系统中,利用 Spark streamimg 或者 Flink 对接数据流并进行数据分析是一种常见请求。常见的场景如下:大促开始后,实时显示当前总成交金额。大促开始后,实时画出成交量走势图。……MySQL 若需要支持此类场景,需要开发应用,解析 binlog 中的数据,对接中间件,开发成本运维成本都会更高,且系统复杂度也会提升。而表格存储提供了通道服务,可以直接对接 Spark stre

背景

在订单系统中,利用 Spark streamimg 或者 Flink 对接数据流并进行数据分析是一种常见请求。常见的场景如下:

  • 大促开始后,实时显示当前总成交金额。

  • 大促开始后,实时画出成交量走势图。

  • ……


MySQL 若需要支持此类场景,需要开发应用,解析 binlog 中的数据,对接中间件,开发成本运维成本都会更高,且系统复杂度也会提升。而表格存储提供了通道服务,可以直接对接 Spark streaming 或 Flink,不仅省去了开发者解析数据、对接中间件的工作,且可以将开发者从复杂的系统架构中解放出来,更加专注于业务逻辑的处理。


本文将一步一步展示,如何利用表格存储通道服务,实现对订单系统实时成交额和订单数的统计工作。

基于通道服务的 Tablestore 解决方案

通道服务说明

通道服务(Tunnel Service)是基于表格存储数据接口上的全增量一体化服务。通道服务提供了增量、全量、增量加全量三种类型的分布式数据实时消费通道。通过为数据表建立数据通道,可以简单地实现对表中历史存量和新增数据的消费处理。具体可参考通道服务概述


我们可以将 Spark streamimg 或者 Flink 和 Tablestore 的通道服务进行对接,对于表格存储中的数据变动,进行实时计算,完成上述场景中的需求。

本文架构

本文分别使用 Spark streaming 和 Flink 对接 Tablestore 的通道服务,完成对订单数量和成交额的实时统计。原始订单数据由 Java 服务写入到 MySQL,再由 DTS 服务将数据同步到 Tablestore,这一部分内容已经在前面的文章中详细说明。Spark streaming / Flink 通过通道服务拿到实时数据变化,聚合,将统计结果写回到 Tablestore 中的 sink 表中。完整架构如下:

 

准备工作

Tablestore 申请


创建源表并开通 Tunnel 服务

在 Tablestore 中建表的过程不再进行描述。本文,使用 DTS 同步工具将 MySQL 中的数据同步到 Tablestore 中,Tablestore 中的订单表 order_contract 为 DTS 服务自动创建,其表结构如下:

列名称

类型

列说明

字段内容说明

oId

STRING

主键

订单id

c_id

STRING


客户id

c_name

STRING


客户名称

create_time

STRING


订单创建时间

has_paid

INTEGER


是否已经支付

p_brand

STRING


商品品牌

p_count

INTEGER


商品数量

p_id

STRING


商品id

p_name

STRING


商品名称

p_price

DOUBLE


商品单价

pay_time

STRING


支付时间

s_id

STRING


商家id

s_name

STRING


商家名称

total_price

DOUBLE


订单金额


建表后在实例管理页,找到“数据表列表”,点击刚刚创建的表order_contract进入表管理页面。

在表管理页面,选择实时消费通道创建通道

通道名称自定,通道类型选择增量。点击确定完成通道创建。

 

此时可以看到创建的通道的通道 id。

创建 sink 表

sink 表用于存储 Spark streaming 计算后的结果数据。在本文中 sink 表命名为 order_sink,用于存储单位时间段内订单成交数量和成交额。其表结构如下:

列名称

类型

列说明

字段内容说明

order_start

STRING

主键

记录时间段的开始时间

order_end

STRING


记录时间段的结束时间

order_count

INTEGER


记录时间范围内的总订单数

total_price

DOUBLE


记录时间范围内的总成交额

Spark streaming 对接 Tunnel

创建集群

创建阿里云E-MapReduce的Hadoop集群,文档参见创建集群

登录 Spark-sql 客户端

集群管理页面,点击创建的集群。

点击主机列表,点击emr-header-1机器。

点击远程连接

选择立即登录

输入创建集群时设定的密码。登录机器。

在指令行输入以下指令,登录 Spark-sql 客户端,

streaming-sql --driver-class-path emr-datasources_shaded_*.jar --jars emr-datasources_shaded_*.jar --master yarn-client --num-executors 8 --executor-memory 2g --executor-cores 2

进入如下界面。

流计算

在 Spark-sql 中执行 SQL,创建源表,

DROP TABLE IF EXISTS order_contract;
CREATE TABLE order_contract
USING tablestore
OPTIONS(
endpoint="https://test-20210609.cn-hangzhou.vpc.tablestore.aliyuncs.com",
access.key.id="",
access.key.secret="",
instance.name="test-20210609",
table.name="order_contract",
tunnel.id="3d71bb67-58da-4c72-b36f-08b79df7c85d",
catalog='{"columns": {"oId": {"col": "oId", "type": "string"}, "total_price": {"col": "total_price", "type": "double"}, "pay_time": {"cols": "pay_time", "type": "long"}}}'
);

执行以下 SQL 创建目标表,

DROP TABLE IF EXISTS order_sink;
CREATE TABLE order_sink
USING tablestore
OPTIONS(
endpoint="https://test-20210609.cn-hangzhou.vpc.tablestore.aliyuncs.com",
access.key.id="",
access.key.secret="",
instance.name="test-20210609",
table.name="order_sink",
catalog='{"columns":{"order_start":{"col":"order_start","type":"string"},"order_end":{"col":"order_end","type":"string"},"order_count":{"col":"order_count","type":"long"},"total_price":{"total_price":"end","type":"double"}}}'
);

在目标表建立视图,

CREATE SCAN order_contract_view ON order_contract USING STREAM OPTIONS ("maxoffsetsperchannel"="10000");

创建 Stream 任务,

CREATE STREAM job1
options(
checkpointLocation='/tmp/spark/cp/job1',
outputMode='update'
)
INSERT INTO order_sink
SELECT CAST(window.start AS String) AS order_start, CAST(window.end AS String) AS order_end, 
count(*) AS order_count, sum(total_price) AS total_price 
FROM order_contract_view 
GROUP BY window(to_timestamp(pay_time / 1000000), "30 seconds");

提交任务后,可以看到机器执行流式任务输出的日志。

计算结果

可以在 order_sink 表中看到实时统计结果如图。

Flink 对接 Tunnel

创建集群

下单后,在控制台,点击创建集群

选择订单号,点击下一步。

填入集群信息。

选择 OSS Bucket、VPC、Zone。若没有可以点击右侧问号新建。点击下一步,点击创建。

等待集群启动,启动成功后,可以点击创建项目

创建项目

点击创建项目,填入参数。点击确定。

可以在项目列表看到新建的项目。

流计算

在项目列表点击新建的项目,点击开发。

新建作业。输入作业名。

创建作业成功后,在页面中填入如下 SQL。

CREATE TABLE order_contract (
    oId VARCHAR,
    c_id VARCHAR,
    c_name VARCHAR,
    create_time VARCHAR,
    has_paid BIGINT,
    p_brand VARCHAR,
    p_count BIGINT,
    p_id VARCHAR,
    p_name VARCHAR,
    p_price DOUBLE,
    pay_time BIGINT,
    s_id VARCHAR,
    s_name VARCHAR,
    total_price DOUBLE,
    ts AS PROCTIME(),
    primary key(oId)
) WITH (
    type = 'ots',
    instanceName = 'test-20210609',
    tableName = 'order_contract',
    accessId = '',
    accessKey = '',
    endPoint = 'https://test-20210609.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    tunnelName = 'test20210610'
);


CREATE TABLE order_sink (
    order_start VARCHAR,
    order_end VARCHAR,
    order_count BIGINT,
    total_price DOUBLE,
    primary key(order_start)
) WITH (
    type = 'ots',
    instanceName = 'test-20210609',
    tableName = 'order_sink',
    accessId = '',
    accessKey = '',
    endPoint = 'https://test-20210609.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    valueColumns = 'order_end,order_count,total_price'
);

INSERT INTO order_sink
SELECT 
    DATE_FORMAT(TUMBLE_START(order_contract.ts, INTERVAL '30' SECOND), 'yyyy-MM-dd hh:mm:ss') AS order_start,
    DATE_FORMAT(TUMBLE_END(order_contract.ts, INTERVAL '30' SECOND), 'yyyy-MM-dd hh:mm:ss') AS order_end,
    COUNT(oId) as order_count,
    SUM(total_price) as total_price
FROM order_contract
GROUP BY TUMBLE(ts, INTERVAL '30' SECOND);

点击语法检查,通过后。点击上线

选择系统分配,点击下一步继续。

SQL 检查通过后,继续点击下一步。填入期望事件后,点击下一步,点击上线。

上线成功后,可以在运维页看到作业如图。

计算结果

可以在 order_sink 表中看到实时统计结果如图。

总结

表格存储通过通道服务,对流式数据计算进行了支持。相比于 MySQL 中解析 binlog 方案,使用 Tablestore 通道服务直接对接流式处理工具,运维开发成本更低,系统架构更加简单。本文分别使用了 Spark streaming 和 Flink 对订单系统中典型场景进行了模拟操作,完成了实时统计交易额、订单数的流式作业,整个过程简单易懂。

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
2月前
|
安全 关系型数据库 MySQL
如何将数据从MySQL同步到其他系统
【10月更文挑战第17天】如何将数据从MySQL同步到其他系统
198 0
|
1月前
|
运维 Serverless 数据处理
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
80 1
|
1月前
|
关系型数据库 MySQL Linux
Linux系统如何设置自启动服务在MySQL数据库启动后执行?
【10月更文挑战第25天】Linux系统如何设置自启动服务在MySQL数据库启动后执行?
83 3
|
1月前
|
运维 监控 Serverless
Serverless架构在图像处理等计算密集型应用中展现了显著的优势
Serverless架构在图像处理等计算密集型应用中展现了显著的优势
31 1
|
2月前
|
运维 Serverless 数据处理
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
Serverless架构通过提供更快的研发交付速度、降低成本、简化运维、优化资源利用、提供自动扩展能力、支持实时数据处理和快速原型开发等优势,为图像处理等计算密集型应用提供了一个高效、灵活且成本效益高的解决方案。
61 3
|
2月前
|
存储 固态存储 安全
阿里云服务器X86计算架构解析与X86计算架构云服务器收费价格参考
阿里云服务器架构分为X86计算、Arm计算、高性能计算等多种架构,其中X86计算是用户选择最多的一种架构,本文将深入探讨阿里云X86计算架构的云服务器,包括其技术特性、适用场景、性能优势以及最新价格情况。
|
2月前
|
编解码 弹性计算 应用服务中间件
阿里云服务器Arm计算架构解析:Arm计算架构云服务器租用收费标准价格参考
阿里云服务器架构分为X86计算、Arm计算、高性能计算等多种架构,其中Arm计算架构以其低功耗、高效率的特点受到广泛关注。本文将深入解析阿里云Arm计算架构云服务器的技术特点、适用场景以及包年包月与按量付费的收费标准与最新活动价格情况,以供选择参考。
|
2月前
|
存储 关系型数据库 MySQL
PACS系统 中 dicom 文件在mysql 8.0 数据库中的 存储和读取(pydicom 库使用)
PACS系统 中 dicom 文件在mysql 8.0 数据库中的 存储和读取(pydicom 库使用)
36 2
|
2月前
|
Ubuntu 关系型数据库 MySQL
Linux系统MySQL安装
【10月更文挑战第19天】本文介绍了在 Linux 系统上安装 MySQL 的步骤,包括安装前准备、安装 MySQL、启动 MySQL 服务、配置 MySQL 以及验证安装。适用于 Ubuntu/Debian 和 CentOS/Fedora 系统,提供了详细的命令示例。
212 1
|
存储 索引
表格存储根据多元索引查询条件直接更新数据
表格存储是否可以根据多元索引查询条件直接更新数据?
114 3