使用 Amazon Redshift 构建用于批量和实时分析的大数据 Lambda 架构

通过掌握有关客户、产品和应用程序的实时信息，组织可以在业务应用程序中发生事件时采取行动。例如，您可以防止财务欺诈、提供个性化内容，以及近实时地识别和预防故障。尽管批量分析提供了大规模分析趋势和处理数据的功能，允许在时间间隔内处理数据（例如各商店的每日销售汇总），但实时分析针对低延迟分析进行了优化，可确保数据在几秒钟内可供查询。这两种数据处理范式都在孤岛中运行，这会导致数据冗余和维护它们的运营开销。大数据 Lambda 架构是一种参考架构模式，它允许用于分析的大规模数据的批处理和近实时范式无缝共存。

借助 Amazon Redshift，您可以使用标准 SQL 轻松分析数据仓库、运营数据库和数据湖中的所有数据类型。在这篇博文中，我们将实时收集、处理和分析数据流。通过数据共享，您可以在 Amazon Redshift 集群之间共享实时数据以进行读取，同时具有相对的安全性和开箱即用性。在这篇博文中，我们将讨论如何利用 Amazon Redshift 的数据共享功能来设置大数据 Lambda 架构，以支持批量分析和近实时分析。

解决方案概览

Example Corp. 是一家领先的电动汽车公司，彻底改变了汽车行业。Example Corp. 对联网车辆数据进行操作，并提高各种联网车辆和车队用例的有效性，包括预测性维护、车载服务变现、基于使用情况的保险，以及提供卓越的驾驶员体验。在这篇博文中，我们使用联网汽车数据探讨了实时和趋势分析，以说明以下用例：

基于使用情况的保险 — 基于使用情况的保险 (UBI) 依赖于对驾驶员车辆近实时数据的分析，以了解驾驶员的风险状况。此外，它还依赖于指标的历史分析（批量）（例如一年内行驶的里程数）。司机越好，保费越低。
车队绩效趋势 — 车队（例如出租车车队）的绩效依赖于对整个车队（批量）数据历史趋势的分析，以及向下钻取到车队中的一辆车以近实时分析油耗或驾驶员分心等指标的能力。

架构概览

在本部分，我们将讨论 Lambda 架构解决方案的整体架构设置。

下图显示了实现架构和不同的计算层：

从 AWS IoT Core 摄取数据
批处理层
速度层
服务层

数据摄取

车辆遥测数据通过 AWS IoT Core 摄取到云中，然后路由到 Amazon Kinesis Data Streams。Kinesis Data Streams 层充当速度层和批处理层的分离层，其中传入的遥测数据分别由速度层的 Amazon Redshift 集群和 Amazon Kinesis Data Firehose 使用。

批处理层

Amazon Kinesis Data Firehose 是一项完全托管式服务，可以在将数据流加载到 Amazon Simple Storage Service (Amazon S3) 数据湖之前对其进行批处理、压缩、转换和加密。Kinesis Data Firehose 还允许您为交付数据记录的 Amazon S3 前缀指定自定义表达式。这样就可以筛选分区数据并控制每个查询扫描的数据量，从而提高性能并降低成本。

批处理层将数据保存在 Amazon S3 中，并由 Amazon Redshift Serverless 端点（服务层）直接访问。使用 Amazon Redshift Serverless，您可以对来自 S3 内文件的结构化和半结构化数据进行高效查询与检索，而不用将数据加载到 Amazon Redshift 表中。

批处理层还可以选择将结果预计算为来自不可变 Amazon S3 数据湖的批处理视图，并将它们保存为原生表或实例化视图，以用于高性能用例。您可以使用 AWS Glue、Amazon Redshift 存储过程、Amazon Redshift 实例化视图或其他选项创建这些预先计算的批处理视图。

批处理视图的计算公式为：

批处理视图 = 函数（所有数据）

在此解决方案中，我们为 Example Corp. 构建了用于两种查询的批处理层：

rapid_acceleration_by_year — 每位驾驶员每年累计的快速加速次数
total_miles_driven_by_year — 车队每年累计行驶的里程总数

出于演示目的，我们使用 Amazon Redshift 存储过程从使用 Amazon Redshift Spectrum 的外部表创建批处理视图作为 Amazon Redshift 原生表。

速度层

速度层实时处理数据流，旨在通过提供最新数据的实时视图来最大限度地减少延迟。

Amazon Redshift Streaming Ingestion 使用 SQL 同时连接一个或多个 Kinesis 数据流。借助 Amazon Redshift 中的原生流式摄取功能，您可以直接从 Kinesis Data Streams 中摄取数据，并允许您每秒摄取数百兆字节的数据，同时以极低的延迟进行查询（在许多情况下，在进入数据流后仅需 10 秒即可完成）。

速度集群使用实例化视图来示例 Kinesis 数据流的时间点视图，该视图在查询时累积。实时视图使用此层进行计算，此层提供了传入遥测流的近实时视图。

速度视图可以作为批处理视图中未考虑的最近数据的函数来计算：

速度视图 = 函数（最近的数据）

我们按如下方式计算这些批处理视图的速度视图：

rapid_acceleration_realtime — 在批处理视图 rapid_acceleration_by_month 中未考虑到的最近数据的每位司机的快速加速次数
miles_driven_realtime — 不是以 miles_driven_by_month 为单位的最近数据的每位司机驾驶的里程数

服务层

服务层包括一个 Amazon Redshift Serverless 端点和任何消耗服务，例如 Amazon QuickSight 或 Amazon SageMaker。

Amazon Redshift Serverless（预览版）是 Amazon Redshift 的无服务器选项，它可以在几秒钟内轻松运行和扩展分析，而无需设置和管理数据仓库基础设施。借助 Amazon Redshift Serverless，任何用户（包括数据分析师、开发人员、业务专业人员和数据科学家）只需在数据仓库中加载和查询数据即可获得数据见解。

Amazon Redshift 数据共享支持在 Amazon Redshift 集群之间进行即时、精细和快速的数据访问，而无需维护数据的冗余副本。

速度集群向 Amazon Redshift Serverless 端点（服务集群）提供实时实例化视图的出站数据共享。

服务集群联接来自批处理层和速度层的数据，以最小的延迟获取特定函数的近实时和历史数据。消耗层（如 Amazon API Gateway 或 QuickSight）只知道服务集群，所有的批处理和流处理都从消耗层抽象而来。

我们可以按如下所示从数据消耗层查看速度层的查询：

查询 = 函数（批处理视图、速度视图）

部署 CloudFormation 模板

我们提供了一个 AWS CloudFormation 模板来演示该解决方案。您可以下载并使用此模板轻松部署所需的 AWS 资源。此模板已在 us-east-1 区域进行了测试。

此模板要求您提供以下参数：

DatabaseName — 要为速度集群创建的第一个数据库的名称
NumberOfNodes — 集群中的计算节点数量。
NodeType — 要预置的节点类型
MasterUserName — 与正在创建的集群的主用户账户关联的用户名
MasterUserPassword — 与主用户账户关联的密码
InboundTraffic — 允许集群入站流量的 CIDR 范围
PortNumber — 集群接受传入连接的端口号
SQLForData — 要从 AWS IOT Core 主题中提取的源查询

先决条件

设置此解决方案并使用自己的应用程序数据推送到 Kinesis Data Streams 时，可以跳过设置 IoT 设备模拟器，并开始创建 Amazon Redshift Serverless 端点。这篇博文使用模拟器来创建相关的数据库对象，并假定在解决方案演练中使用模拟器。

设置 IoT 设备模拟器

我们使用 IoT 设备模拟器来生成和模拟车辆 IoT 数据。该解决方案允许您创建和模拟数百个联网设备，而无需配置和管理物理设备或开发耗时的脚本。

使用以下 CloudFormation 模板在您的账户中创建 IoT 设备模拟器以试用此解决方案。

配置设备和模拟

要配置设备和模拟，请完成以下步骤：

使用您在提供的电子邮件中收到的登录信息登录 IoT 设备模拟器。
选择 Device Types（设备类型）和 Add Device Type（添加设备类型）。
选择 Automotive Demo（汽车演示）。
对于 Device type name（设备类型名称），输入 testVehicles。
对于 Topic（主题），输入将传感器数据发送到 AWS IoT Core 的主题。
保存设置。
选择 Simulations（模拟）和 Add simulation（添加模拟）。
对于 Simulation name（模拟名称），输入 testSimulation。
对于 Simulation type（模拟类型），选择 Automotive Demo（汽车演示）。
对于 Select a device type（选择设备类型），选择您创建的设备类型（testVehicles）。
对于 Number of devices（设备数量），输入 15。

每次模拟最多可以选择 100 台设备。您可以配置更多数量的设备来模拟大数据。

对于 Data transmission interval（数据传输间隔），输入 1。
对于 Data transmission duration（数据传输持续时间），输入 300。

此配置将运行模拟 5 分钟。

选择 Save（保存）。

现在，您已准备好向 AWS IoT Core 模拟车辆遥测数据。

创建 Amazon Redshift Serverless 端点

该解决方案使用 Amazon Redshift Serverless 端点作为服务层集群。您可以在自己的账户中设置 Amazon Redshift Serverless。

设置 Amazon Redshift 查询编辑器 V2

要查询数据，您可以使用 Amazon Redshift 查询编辑器 V2。有关更多信息，请参阅隆重推出 Amazon Redshift 查询编辑器 V2，这是一款基于 Web 的免费查询编写工具，专为数据分析师设计。

获取预置速度层集群和 Amazon Redshift Serverless 的命名空间

通过查询编辑器 V2 连接到 speed-cluster-iot（速度层集群），然后运行以下 SQL：

select current_namespace; -- (Save as <producer_namespace>)

同样，连接到 Amazon Redshift Serverless 端点并获取命名空间：

select current_namespace; -- (Save as <consumer_namespace>)

您也可以通过 Amazon Redshift 控制台获取此信息。

现在我们已经设置好了所有先决条件，接下来我们来完成解决方案演练。

实施解决方案

该工作流包括以下步骤：

启动在上一部分中创建的 IoT 模拟。

车辆 IoT 通过 IoT 设备模拟器模拟和摄取，用于配置的车辆数量。原始遥测负载被发送到 AWS IoT Core，AWS IoT Core 将数据路由到 Kinesis Data Streams。

在批处理层，数据直接从 Kinesis Data Streams 放入 Kinesis Data Firehose，Kinesis Data Firehose 会将数据转换为 parquet，然后以前缀 s3://<Bucketname>/vehicle_telematics_raw/year=<>/month=<>/day=<>/ 发送到亚马逊。

模拟完成后，在 AWS Glue 控制台上运行预先创建的 AWS Glue 爬网程序 vehicle_iot_crawler。

服务层 Amazon Redshift Serverless 端点可以通过 Redshift Spectrum 外部表直接访问来自 Amazon S3 数据湖的数据。在本演示中，我们通过 Redshift Spectrum 计算批处理视图，然后使用 Amazon Redshift 存储过程将它们存储为 Amazon Redshift 表。

通过查询编辑器 V2 连接到 Amazon Redshift Serverless 端点，并使用以下 SQL 脚本创建存储过程。
运行两个存储过程来创建批处理视图：

call rapid_acceleration_by_year_sp();
call total_miles_driven_by_year_sp();

这两个存储过程将批处理视图创建为 Amazon Redshift 原生表：

- batchlayer_rapid_acceleration_by_year
- batchlayer_total_miles_by_year

您还可以将这些存储过程安排为批处理作业。有关更多信息，请参阅在您的 Amazon Redshift 数据仓库上安排 SQL 查询。

在速度层，传入的数据流由实例化视图 vehicleiotstream_mv 中的速度层 Amazon Redshift 集群读取和实例化。

连接到预置的 speed-cluster-iot，然后运行以下 SQL 脚本来创建所需的对象。

从此实例化视图创建了两个实时视图：

- batchlayer_rapid_acceleration_by_year
- batchlayer_total_miles_by_year

按所需的间隔刷新实例化视图 vehicleiotstream_mv，这将触发 Amazon Redshift 从流中读取数据并将数据加载到实例化视图中。
```
REFRESH MATERIALIZED VIEW vehicleiotstream_mv;
```

刷新目前是手动的，但可以使用查询计划程序自动刷新。

实时视图作为出站数据共享由速度集群共享到服务集群。

连接到 speed-cluster-iot 并使用以下 SQL 创建出站数据共享（生产者）：

-- Create Datashare from Primary (Producer) to Serverless (Consumer)
CREATE DATASHARE speedlayer_datashare SET PUBLICACCESSIBLE TRUE;
ALTER DATASHARE speedlayer_datashare ADD SCHEMA public;
ALTER DATASHARE speedlayer_datashare ADD ALL TABLES IN SCHEMA public;
GRANT USAGE ON DATASHARE speedlayer_datashare TO NAMESPACE '<consumer_namespace>'; --（替换为先决条件 5 中创建的使用者命名空间）

连接到 speed-cluster-iot 并使用以下 SQL 创建入站数据共享（使用者）：

CREATE DATABASE vehicleiot_shareddb FROM DATASHARE speedlayer_datashare OF NAMESPACE '< producer_namespace >'; --（替换为先决条件 5 中创建的生产者命名空间）

现在 Amazon Redshift Serverless 端点的实时视图已经可用，我们可以运行查询，通过访问批处理层和速度层并使用以下查询将它们联接起来，来获取实时指标或历史趋势以及最新数据。

例如，要使用最新数据计算每年的总快速加速，可以运行以下查询：

-- Rapid Acceleration By Year

select SUM(rapid_acceleration) rapid_acceleration, vin, year from 
(
select rapid_acceleration, vin,year
  from public.batchlayer_rapid_acceleration_by_year batch
union all
select rapid_acceleration, vin,year
from speedlayer_shareddb.public.speedlayer_rapid_acceleration_by_year speed)
group by VIN, year;

同样，要使用最新数据计算每年行驶的总里程数，请运行以下查询：

-- Total Miles Driven By Year

select SUM(total_miles) total_miles_driven , year from 
(
select total_miles, year
  from public.batchlayer_total_miles_by_year batch
union all
select total_miles, year
from speedlayer_shareddb.public.speedlayer_total_miles_by_year speed)
group by year;

为了仅访问实时数据以支持日常控制面板，您可以针对共享到 Amazon Redshift Serverless 集群的实时视图运行查询。

例如，要计算车队每次行程的平均速度，可以运行以下 SQL：

select CAST(measuretime as DATE) "date",
vin,
trip_id,
avg(vehicleSpeed)
from speedlayer_shareddb.public.vehicleiotstream_mv 
group by vin, date, trip_id;

由于本演示使用了与快速入门相同的数据，因此本演示中有重复项。在实际实施中，服务集群通过创建具有日期谓词的视图来管理数据冗余和重复项，这些数据谓词使用批处理和实时视图中的非重叠数据，并向消耗层提供总体指标。

您可以通过 QuickSight 将数据用于控制面板，使用 API Gateway 进行基于 API 的访问，或者通过 Amazon Redshift Data API 或 SageMaker 支持 AI 和机器学习 (ML) 工作负载。这不包含在提供的 CloudFormation 模板中。

最佳实践

在本部分，我们将讨论使用此解决方案时的一些最佳实践和经验教训。

预置与无服务器

速度层是一个连续摄取层，从通常运行全天候工作负载的 IoT 流中读取数据。工作负载的空闲时间和可变性更少，拥有支持可弹性扩展的持久性工作负载的预置集群是有利的。

对于零星或临时工作负载，可以预置服务层（如果是全天候工作负载）或 Amazon Redshift Serverless。在这篇博文中，我们假设是零星工作负载，因此预置 Serverless 最合适。此外，服务层可以容纳多个 Amazon Redshift 集群，每个集群都使用各自的数据共享并为下游应用程序提供服务。

用于数据共享的 RA3 实例

Amazon Redshift RA3 实例支持数据共享，使您能够安全、轻松地在 Amazon Redshift 集群之间共享实时数据以供读取。您可以使用数据共享将近实时摄取的数据与历史数据相结合，以提供个性化的驾驶特征，从而确定保险建议。

您还可以根据需要将生产者中基础数据的细粒度访问控制授予使用者集群。Amazon Redshift 使用系统表和 AWS CloudTrail 提供全面的审计功能，允许您监控所有使用者的数据共享权限和使用情况，并在必要时立即撤销访问权限。这些权限由生产者和使用者集群的超级用户授予，用于定义谁可以访问哪些对象，这与前面部分中使用的授权命令类似。您可以使用以下命令来审核数据共享的使用情况和活动。

使用以下代码跟踪对数据共享和从数据共享导入的共享数据库的所有更改：

Select username, share_name, recordtime, action,
         share_object_type, share_object_name 
  from svl_datashare_change_log
   order by recordtime desc;

使用以下代码跟踪仅与生产者相关的数据共享访问活动（使用情况）：

Select * from svl_datashare_usage;

暂停和恢复

您可以在批处理完成后暂停生产者集群以节省成本。借助 Amazon Redshift 上的暂停和恢复操作，您可以轻松地暂停和恢复可能并非始终处于运行状态的集群。它允许您创建一个定期安排的时间，以便在特定时间启动暂停和恢复操作，也可以手动启动暂停和稍后恢复。灵活的按需定价和按秒计费让您可以更好地控制 Redshift 计算集群的成本，同时以易于管理的方式维护数据。

用于快速访问数据的实例化视图

实例化视图允许对大型表进行复杂查询的预合成结果，以便更快地访问。生产者集群将数据公开为实例化视图，以简化使用者集群的访问。这也允许生产者集群灵活地更新基础表结构以适应新的业务用例，而不会影响依赖使用者的查询，也不会启用松耦合。

结论

在这篇博文中，我们演示了如何在 Lambda 架构原则的指导下，使用 Amazon Redshift 作为平台的核心，处理和分析来自流媒体和批处理源的大规模数据。

您首先从联网车辆收集实时数据，然后通过 Kinesis Data Firehose 将流数据存储在 Amazon S3 数据湖中。该解决方案可同时处理数据，以便通过 Amazon Redshift 流媒体摄取进行近实时的分析。

通过数据共享功能，您可以将实时的最新数据共享到 Amazon Redshift Serverless 端点（服务集群），该端点将速度层（近实时）和批处理层（批量分析）中的数据进行合并，从而提供对从近实时分析到历史趋势数据的低延迟访问。

单击此处立即开始使用此解决方案，并通过评论部分告诉我们您如何在自己的组织中实施此解决方案。

关于作者

Jagadish Kumar 是 AWS 的高级分析专家解决方案架构师。他对数据架构充满热情，并帮助客户在 AWS 上大规模构建分析解决方案。他是一名狂热的大学橄榄球迷，喜欢阅读、看体育赛事和骑摩托车。

Thiyagarajan Arumugam 是 Amazon Web Services 的一名大数据解决方案架构师，负责帮助客户设计大规模处理数据的架构。在加入 AWS 之前，他为 Amazon.com 构建了各种数据仓库解决方案。在闲暇时间，他喜欢所有户外运动，喜欢练习弹奏印度古典鼓 – 魔力单根鼓。

Eesha Kumar 是 AWS 的分析解决方案架构师。他与客户合作，帮助他们利用 AWS 平台和工具构建解决方案，从而实现数据的商业价值。

亚马逊AWS官方博客