Amazon Redshift：`ActiveStatementsExceededException`（如何同时执行 INSERT）

Question

我有一个 Kinesis 集群，它通过 Lambda 将数据推送到 Amazon Redshift。

目前我的 lambda 代码如下所示：

client = boto3.client('redshift-data')
for tx in txs:
    query = # prepare an INSERT query here

    resp = client.execute_statement(
        ClusterIdentifier=redshift_cluster_id,
        Database=redshift_db,
        DbUser=redshift_user,
        Sql=query
    )

问题在于，一旦我尝试扩大 kinesis（更多分片）或 lambda（从单个分片并发处理） - 我就会明白：

[ERROR] ActiveStatementsExceededException: An error occurred (ActiveStatementsExceededException) when calling the ExecuteStatement operation: Active statements exceeded the allowed quota (200).
Traceback (most recent call last):
  File "/opt/python/lib/python3.8/site-packages/codeguru_profiler_agent/aws_lambda/profiler_decorator.py", line 52, in profiler_decorate
    return function(event, context)
  File "/opt/python/lib/python3.8/site-packages/codeguru_profiler_agent/aws_lambda/lambda_handler.py", line 91, in call_handler
    return handler_function(event, context)
  File "/var/task/lambda_function.py", line 71, in lambda_handler
    resp = client.execute_statement(
  File "/var/runtime/botocore/client.py", line 386, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/var/runtime/botocore/client.py", line 705, in _make_api_call
    raise error_class(parsed_response, operation_name)

从 AWS 文档中，我收集到这意味着我正在尝试并行运行太多 execute_statements。

我该如何解决这个问题？使用 Redshift 的唯一方法是通过批处理记录并将它们全部插入在一起吗？

Answer 1

您的代码中的注释让我暂停 - “query = # 在此处准备一个 INSERT 查询”。这似乎意味着您正在将 S3 数据读入 Lambda 并将这些数据插入 Redshift。如果是这样，这不是一个好的模式。

首先，Redshift 期望通过 COPY（或 Spectrum 或 ...）而不是通过 INSERT 将数据引入集群。这将在 Redshift 中产生管理事务的问题，并造成巨大的浪费或磁盘空间/对 VACUUM 的需求。将数据放入 Redshift 的 INSERT 方法是一种反模式，即使是中等大小的数据也不应该这样做。

更普遍的担忧是数据移动阻抗不匹配。 Kinesis 是大量独立的数据流和生成小文件的代码。 Redshift 是一个大型数据库，适用于大型数据段。以一种错过设计目标的方式不匹配这些工具将使它们中的任何一个性能都很差。您需要通过将 S3 批处理到 Redshift 来匹配数据要求。这意味着在单个 COPY 命令中将许多 S3 文件复制到 Redshift。这可以通过清单或 S3 中的“目录”结构来完成。 “从 S3 路径复制所有内容......”这个将数据复制到 Redshift 的过程可以在每个时间间隔（2 或 5 或 10 分钟）运行一次。因此，您希望 Kinesis Lambda 在 S3 中组织数据（或添加到清单），以便可以收集“批次”的 S3 文件以执行 COPY。这样可以一次将大量 S3 文件（它的首选数据大小）带入 Redshift，并且还可以大大减少您的执行 API 调用。

现在，如果您设置了一个非常大的 Kinesis 管道并且数据非常大，则需要考虑另一个数据移动“偏好”。这仅在您每分钟移动大量数据时才重要。这种额外的偏好适用于 S3。 S3 作为对象存储意味着“查找”请求的对象键会占用大量时间。大约 0.5 秒。因此读取一千个 S3 对象将需要 500 个（总共）500 秒的键查找时间。 Redshift 将并行向 S3 发出请求，集群中的每个切片一个，因此其中一些时间是并行的。如果正在读取的文件大小为 1KB，则在 S3 查找完成后，数据的数据传输大约需要 1.25 秒。全部的。这一次也是并行的，但您可以看到查找与传输花费了多少时间。为了从 S3 中获得最大带宽以读取许多文件，这些文件的大小需要为 1GB（根据我的经验，100MB 是可以的）。您可以查看是否要每分钟将数百万个文件从 Kinesis 提取到 Redshift 中，您将需要一个流程来将许多小文件组合成更大的文件，以避免 S3 的这种危险。由于您使用 Lambda 作为 Kinesis 阅读器，我希望您还没有达到这个数据速率，但是如果您希望扩展到非常大的规模，那么关注这个问题是件好事。

仅仅因为工具具有高带宽并不意味着它们可以通过管道连接在一起。带宽有多种样式。