如何在写入 Redshift DW 之前从 S3 存储桶转换数据？

Posted 2023-03-29

【中文标题】如何在写入 Redshift DW 之前从 S3 存储桶转换数据？

【英文标题】：How to transform data from S3 bucket before writing to Redshift DW?

【发布时间】：2020-02-11 21:36:25

【问题描述】：

我正在用 redshift 创建一个（现代）数据仓库。我们所有的基础设施都托管在亚马逊上。到目前为止，我已经设置 DMS 以从我们业务数据库的某些表（EC2 上的 SQL Server，而不是 RDS）中提取数据（包括更改的数据）并将其直接存储到 S3。

现在，我必须先转换和丰富 S3 中的这些数据，然后才能将其写入 Redshift。我们的 DW 有一些事实和维度表（星型模式），所以，想象一个客户维度，它不仅应该包含客户基本信息，还应该包含地址信息、城市、州等。这些数据分布在几个表中我们的业务数据库。

所以这是我的问题，我不清楚如何查询 S3 暂存区以连接这些表并将其写入我的 redshift DW。我想使用 Glue、Kinesis 等 AWS 服务，即完全无服务器。

Kinesis 能否完成这项任务？如果我将暂存区从 S3 移到 Redshift 会不会让事情变得更容易，因为我们所有的数据本质上都是高度相关的？如果是这样，问题仍然存在，如何在将数据保存到我们的 DW 模式之前对其进行转换/丰富？我到处搜索这个特定的主题，但是关于它的信息很少。

感谢任何帮助。

【参考方案1】：

有很多方法可以做到这一点，但一种想法是使用 Redshift Spectrum 查询数据。 Spectrum 是一种使用 Redshift 集群查询 S3（称为外部数据库）的方法。

真正高级别的方法是创建一个 Glue Crawler 作业来抓取您的 S3 存储桶，这会创建 Redshift Spectrum 可以查询的外部数据库。

这样，您无需将数据移动到 Redshift 本身。您可能希望将“暂存”区域保留在 S3 中，并且只将准备好用于报告或分析的数据（即 Customer Dim 表）引入 Redshift。

这是执行此操作的文档：https://docs.aws.amazon.com/redshift/latest/dg/c-getting-started-using-spectrum.html

要调度 ETL SQL：我不相信 Redshift 中内置了调度工具，但您可以通过以下几种方式做到这一点：

1) 获取 ETL 工具或在服务器或 Glue 上设置 CRON 作业，以计划运行 SQL 脚本。我使用连接到数据库的 Python 脚本执行此操作，然后运行 ​​SQL 文本。这将是一个多一点的批量操作。您也可以在 Lambda 函数中执行此操作，并将其安排在 Cloudwatch 触发器上，该触发器可以在 cron 计划上

2) 使用 Lambda 函数运行您希望触发 S3 PUT 到该存储桶的 SQL 脚本。这样脚本将在文件删除时正确运行。这基本上是一个实时操作。 DMS 会非常快速地删除文件，因此您将每分钟多次删除文件，这可能更难以处理。

【讨论】：

好主意！实际上，S3 保留了“原始”表，它们可以在通过INSERT INTO 或CREATE TABLE AS 加载到 Redshift 表中时进行转换。

你所描述的正是我想要做的。如果您愿意，我想将原始数据保留在 S3 存储桶中，就像数据湖一样，然后将数据带到 Redshift 以提供我的维度和事实。回到你的解决方案，你能更具体一点吗？比如，如何触发频谱来查询 S3 并更新我的表？每次我在 S3 上收到新数据时，我都需要它运行。

我刚刚在我的回答中添加了更多细节

我想我会尝试第二种方式。当文件写入存储桶时触发 lambda 函数对我来说是完美的。我想知道nodejs中的一个函数是否足够......我对python不熟悉。

另外，我认为我可以将 DMS 数据发送到 kinesis 流而不是直接发送到存储桶，这样我可以设置流放置文件的时间间隔。无论如何，非常感谢您的帮助！

【参考方案2】：

一种选择是将“原始”数据作为“暂存”表加载到 Redshift。然后，运行 SQL 命令将数据（JOIN 等）操作为所需的格式。

最后，将生成的数据复制到用户查询的“公共”表中。

这是一个普通的Extract-Load-Transform过程（与 ETL 略有不同），它使用 Redshift 的功能进行转换。

【讨论】：

是的，但是我怎样才能自动完成这项任务呢？每次暂存区的数据发生变化时，我如何触发一个任务来转换数据并将其复制到公共表中？我可以使用触发器来做到这一点，但是，AFAIK，Redshift 没有触发器。

您通过 JDBC 或 ODBC 连接与 Redshift 进行通信。例如，Python 应用程序可以使用psycopg2。该应用程序可以连接到 Redshift，然后向其发送命令以运行。或者，您可以使用psql 命令行工具（Redshift 基于 PostgreSQL）并为其提供要运行的命令脚本。当您希望加载数据时，您会触发此类脚本——它可以从 S3 加载数据，然后在一个脚本中对其进行转换。