使用 Amazon Athena 的联合查询和由用户定义的函数简化 ETL 数据管道

Original URL:https://thinkwithwp.com/blogs/big-data/simplify-etl-data-pipelines-using-amazon-athenas-federated-queries-and-user-defined-functions/

Amazon Athena 最近在预览版中增加了对联合查询及由用户定义的函数 (UDF) 的支持。请参阅使用 Amazon Athena 的全新联合查询对任何数据源进行查询，以了解更多详细信息。Jornaya 帮助营销人员智能的与市场中购买大型生活用品（如房屋、抵押贷款、汽车、保险和教育等）的消费者建立联结。

Jornaya 从各种数据源收集数据。我们的主要挑战是，清理这些数据并将其提取到 Amazon S3，以使数据分析师和数据科学家可以进行访问。

传统 ETL 和分析解决方案

2012 年，Jornaya 从 MySQL 迁移到 Amazon DynamoDB作为我们的实时数据库。DynamoDB 让我们这种规模的公司可以享受到创建、读取、更新和删除 (CRUD) 操作的好处，并提供可预测的低延迟、高可用性和出色的数据持久性，而且无需管理数据库。这让我们的技术团队能够专注于解决业务问题，并且快速开发可以推到市场的新产品。

在 NoSQL 数据库上运行分析查询有时非常棘手。我们决定从 DynamoDB 提取数据，然后在此类数据上运行查询。但这并不简单。

Jornaya 使用以下几种从 DynamoDB 获取数据的方法：

利用 EMR：我们临时为DynamoDB 表预置额外的读取容量，创建瞬态 EMR 集群，以便从 DynamoDB 读取数据并写入到 Amazon S3。
- 我们的 Jenkins 作业会触发启动集群的管道，通过 EMR 提取数据，并使用 Amazon Redshift copy 命令将数据加载到 Amazon Redshift。这是一个昂贵的流程，因为我们会使用了过多的读取容量。为了降低 EMR 成本，我们开始采用竞价型实例。
启用 DynamoDB 流：我们借助内部开发的名为 Dynahose 的 Python AWS Lambda 函数来使用来自流的数据，然后将其写入到 Amazon Kinesis Firehose delivery stream。然后，我们会配置 Kinesis Firehose delivery stream将数据写入 Amazon S3 位置。最后，我们使用另一项内部开发的名为 Partition 的 Python Lambda 函数以确保与写入 Amazon S3 的数据位置对应的分区已被添加到 AWS Glue 数据目录，使其可以通过 AWS Glue、Amazon Redshift Spectrum、EMR 等工具进行读取。

该流程如下图所示。

我们之所以采用这样的管道，因为我们希望使用 SQL 以自然的方式查询与运营数据有关的问题。

利用 Amazon Athena 简化 ETL 工作流并实现更快的分析

Athena 是一项完全托管的无服务器交互服务，可使用 SQL 对 Amazon S3 中的数据进行查询。我们的组织在多个部门都已快速采用这项服务。在我们的用例中，我们不需要始终在线并等待分析查询的 EMR 集群。Athena 的无服务器性质非常适合我们的用例。在此过程中，我们发现可以使用 Athena 运行提取、转换和加载 (ETL) 作业。

但是，Athena 远不止是一项在 Amazon S3 中查询数据的交互服务。我们认为 Athena 还是稳定、有效、可靠、可扩展和高成本效益的 ETL 工具。调度 SQL 语句的能力，及其对从查询结果创建表 (CTAS) 和 INSERT INTO 语句的支持，帮助我们加速运行 ETL 工作负载。

在 Athena 之前，组织内的业务用户必须依靠工程资源来构建管道。Athena 的推出彻底改变了这种状况。Athena 使软件工程师和数据科学家能够处理原本不可访问，或者需要数据工程师帮助才能使用的数据。

通过Athena新增的联合查询和 UDF 功能，Jornaya 可以将很多不稳定的数据管道替换为 Athena，以便从 DynamoDB 提取与转换数据，并将其写入到 Amazon S3。Jornaya 的产品和工程团队注意到我们的 ETL 失败率降低了。财务部门发现 EMR 和 DynamoDB 的成本都有所下降，随时待命的轮班员工（和他们的家人）也能好好地睡个安稳觉。

例如，一个使用 EMR 的 ETL 管道的构建历史记录看起来可能是这样的（在此图表中，ETL 管道执行历史记录的作业执行 id 为在x 轴上，执行时间在 y 轴上，并以分钟为单位）：

在将此管道迁移到 Athena 并使用联合查询对 DynamoDB 进行查询以后，我们现在可以轻松访问数据源，而在过去使用类似于下方的查询是无法做到的：

CREATE TABLE "__TABLE_NAME__"
WITH (
  external_location = '__S3_LOCATION__'
, format = 'PARQUET'
, orc_compression = 'SNAPPY'
, partitioned_by = ARRAY['create_day']
) AS
SELECT DISTINCT
  d.key.s AS device_id
, CAST(d.created.n AS DECIMAL(14, 4)) AS created
, d.token.s AS token
, c.industry AS industry_code
, CAST(CAST(FROM_UNIXTIME(CAST(d.created.n AS DECIMAL(14, 4))) AS DATE) AS VARCHAR) AS create_day
FROM "rdl"."device_frequency_stream" d
  LEFT OUTER JOIN "lambda::dynamodb"."default"."campaigns" c ON c.key = d.campaign_key
WHERE d.insert_ts BETWEEN TIMESTAMP '__PARTITION_START_DATE__' AND TIMESTAMP '__PARTITION_END_DATE__'
  AND d.created.n >= CAST(CAST(TO_UNIXTIME(DATE '__START_DATE__') AS DECIMAL(14, 4)) AS VARCHAR)
  AND d.created.n < CAST(CAST(TO_UNIXTIME(DATE '__END_DATE__') AS DECIMAL(14, 4)) AS VARCHAR);

我们的流程性能大幅提升，其构建历史记录如下图表所示：

结论

使用一个SQL 查询，我们能够处理来自 DynamoDB 的数据，将此类数据转换成 Parquet，应用 Snappy 压缩，在 AWS Glue 数据目录中创建正确的分区，以及将数据提取到 Amazon S3。我们的 ETL 流程执行时间从数小时缩短到几分钟，成本也得到了大幅降低，而且新的流程更简单而且更可靠。在扩展性方面，使用Athena进行ETL的新流程也是面向未来。如果需要从另一个特定用途的数据存储导入数据集，但该数据存储没有现成的数据源连接器，我们可以简单地使用数据源连接器 SDK 编写属于自己的连接器，并在生产中进行部署，而这只需要一次性投入不到一天时间。

除此以外，Athena 联合查询还让 Jornaya 可以对不仅连接来自不同数据源，甚至来自不同数据范式的数据运行查询！我们可以通过运行一次查询实现无缝链接 NoSQL 数据存储、RDS RDBMS 和 Amazon S3 数据湖的数据。

本博文中的内容和意见属于第三方作者，AWS 不对本博文的内容或准确性负责。

亚马逊AWS官方博客

使用 Amazon Athena 的联合查询和由用户定义的函数简化 ETL 数据管道

传统 ETL 和分析解决方案

利用 Amazon Athena 简化 ETL 工作流并实现更快的分析

结论

本篇作者