智能湖仓2.0实践系列：实时数仓架构搭建

扫描上方二维码立即报名

一起共赴年度科技盛宴！

实时数仓的定义

数据仓库最早由 Bill Inmon （被称为数据仓库之父）提出，是一种信息系统的资料存储理论，此理论强调利用某些特殊资料存储方式，让所包含的资料，特别有利于分析处理，以产生有价值的信息并依此作决策。 一般来说，数据仓库可由关系数据库或专为数据仓库开发的多维度数据库创建，其架构可分为星状及雪花状架构，包含数个维度资料表及事实资料表。数据仓库存在的主要作用是被企业用来回答各种各样的数据问题，比如一些在各个行业都常见的数据指标的输出：日活跃用户、收入等。

在大数据时代， 很多企业的数据仓库已经发展成为拥有海量数据以及成千上万张数据表的大型仓储，每天新增的数据量也已经由几年之前的 MB (Megabyte), GB (Gigabyte)，增长到现在的 TB (Terabyte) 级别，怎样存储、计算这些数据也已经不再是数据领域的工作人员讨论的热点，随之而来的更重要的话题是如何能在海量数据中快速的获取业务人员所关注的数据，怎样将各个来源的数据都组织起来，以更高效的方式来支撑业务能力的提升，让数据也能体现出来它们真正的价值，因此，实时数据仓库应运而生。

实时数据仓库顾名思义就是我们将实时查询和分析数据的能力纳入数据仓库中，可以在数据产生后的秒级、分钟级等（不同业务对于实时的定义不同）就可以拿到这些数据，在业务中发挥它们的价值，以提高相应的业务收益。比如游戏行业中在玩家进入游戏时可以从实时数据仓库中查询出这个玩家的活跃程度以及游戏等级，结合这些信息给玩家提供相符合的礼包等；再比如在电商行业，在一次促销活动开始后，运营人员可以及时的关注到活动的效果并采取合适的措施等。

实时数据仓库区别于离线批处理数据仓库的主要特点是，它运用流式计算相关技术来达到业务对与实时性的要求，比如我们经常听到的 Spark Streaming/Flink/Storm 等。下图是一个支撑了多条业务线的大数据部门当前的离线数据仓库和实时数据仓库的对比。从技术层面来说，目前实时数据仓库常用的有Kafka/Kinesis消息队列，再用 Spark Streaming/Flink 去流式地处理数据，最终提供给数据报表，业务产品，或者作为机器学习相关的业务场景的数据输入。

云原生数据仓库

Amazon Redshift 实践实时数仓

Amazon Redshift 是亚马逊云科技的一款数据仓库产品，它能处理基于列式存储的大数据的分析工作，我们的业务数据仓库也主要基于它来搭建。在亚马逊云科技 2021年的 re:Invent 大会上，Amazon Redshift 也发布了 streaming ingestion 的能力，可以直接在 Amazon Redshift 中通过为 Amazon Kinesis Stream 直接创建一个物化视图，以 SQL 来直接查询 stream 中的数据，数据报表和业务产品都可以准实时地获取到这些数据，这相比于我们之前的架构直接省掉了 Spark Streaming 计算的部分，无论从成本、可维护性上来说都有了大幅度的提升。

接下来我们就来看看基于 Amazon Redshift streaming ingestion 的实时数据仓库是怎么搭建的。

1. 创建一个 IAM role 用来赋予 Redshift 读取 Kinesis Stream 的权限，创建 role 可参考亚马逊云科技的官方文档进行创建，创建完成之后，需要给这个 role 赋予以下权限的 policy，这里因为我们的 Redshift 可以对所有的 Kinesis Stream 里的数据都进行读取，因此赋予了这个 role 对于所有的 Kinesis Stream Resource 的权限，如果您只需要访问所有 Kinesis Stream 里的其中几个，也可以在这个 policy 中进行配置： 

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStreamSummary",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords",
                "kinesis:DescribeStream"
            ],
            "Resource": "arn:aws:kinesis:*:000000000000:stream/*"
        },
        {
            "Sid": "ListStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:ListStreams",
                "kinesis:ListShards"
            ],
            "Resource": "*"
        }
    ]
}

*左滑查看更多

亚马逊云科技的官方文档：

https://docs.aws.amazon.com/zh_cn/redshift/latest/mgmt/authorizing-redshift-service.html

2.接下来我们需要在 Redshift 中创建一个 External Schema ，可以在亚马逊云科技管理控制台中打开你的 Redshift 点击“Query editor”或者“Query editor v2”，提前复制好在第一步创建的 IAM role 的 ARN（在亚马逊云科技管理控制台 IAM 服务中可以看到），在 Query editor 中执行以下 SQL：

CREATE EXTERNAL SCHEMA kpi_events FROM KINESIS
IAM_ROLE 'arn:aws:iam::000000000000:role/redshift-xxxxxxxx-role';

*左滑查看更多

3.创建一个 Kinesis Stream 在 Redshift 中的物化视图，其中 stream 名字以*进行保密处理。

CREATE MATERIALIZED VIEW raw_kpi_events DISTKEY(5) sortkey(1) AS
    SELECT approximatearrivaltimestamp,
    partitionkey,
    shardid,
    sequencenumber,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'_id')::character(36) as ID,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'gameId')::varchar(30) as gameId,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'eventTime')::varchar(20) as eventTime,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'revenue')::DECIMAL(10,2) as revenue,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'productId')::varchar(20)  as productId,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'productName')::varchar(100) as productName,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'timezone')::varchar(30) as timezone,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'appVersion')::varchar(30) as appVersion,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'deviceName')::varchar(30) as deviceName,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'osVersion')::varchar(30) as osVersion,
    json_extract_path_text(from_varbyte(data, 'utf-8'),'userID')::varchar(30) as userID
    FROM kpi_events."*********-stream";
 
-- *********-stream 代表你的Kinesis Data Stream的名字

*左滑查看更多

4.刷新您创建的物化视图。

REFRESH MATERIALIZED VIEW raw_kpi_events;

*左滑查看更多

5.查询物化视图中的数据，在 payload 字段就可以看到 stream 中的数据了。

SELECT * FROM raw_kpi_events LIMIT 100;

payload 中的格式为 JSON，以下为例子：

{
 "_id":"271428af-2254-4b49-a669-583324a973c1",
 "gameId":"48",
 "eventTime":"2022-09-26 15:27:17",
 "revenue":"37.00",
 "productId":"103832",
 "productName":"p-18",
 "timezone":"America/Los_Angeles",
 "appVersion":"2.0.8.1",
 "deviceName":"HUAWEI",
 "osVersion":"10.0.1",
 "userID":"453292"
}

*左滑查看更多

如果你想体验整个过程，可以参考以下亚马逊云科技的文档，可以自行创建一个 Kinesis Data Stream，用 Amazon Kinesis Data Generator 往你的 stream 中传入一些模拟的数据，进行试验。

亚马逊云科技参考文档：

https://docs.aws.amazon.com/zh_cn/redshift/latest/dg/materialized-view-streaming-ingestion-example-station-data.html

Amazon Kinesis Data Generator：

https://awslabs.github.io/amazon-kinesis-data-generator/web/producer.html?

6.在 Redshift 中或者实时数据报表中查看、分析数据。

以上就是我们当前的业务使用 Amazon Redshift streaming ingestion 搭建实时数据仓库的一个过程示例，需要提一下的是目前刷新物化视图这一步需要手动跑，不过也可以把它做成一个 Redshift 的 Schedule Query 让它每分钟或每n分钟跑一次， Schedule Query 也支持 crontab 的格式，如下图所示：

欢迎同样使用了 Redshift 这个功能的小伙伴在评论区交流，大家一起讨论在实时数据仓库这个技术领域的更多话题！

本篇作者

贾婷 

亚马逊云科技快速原型解决方案架构师，致力于帮助客户设计和开发大数据方向的快速原型方案，有游戏、汽车行业的大数据开发经验。

扫描上方二维码

与亚马逊云科技一起【凌云驭势 重塑未来】

听说，点完下面4个按钮

就不会碰到bug了！