我有一个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
)
问题是，只要我尝试扩展kineesis(更多分片)或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_statement。
我如何绕过这个?是唯一的方式与红移工作的批处理记录和插入他们都在一起吗?
代码中的注释让我暂停- &;query = #在这里准备一个INSERT查询&;这似乎意味着您正在将S3数据读入Lambda并将该数据插入Redshift。如果是这样，这不是一个好的模式。
首先，Redshift希望通过COPY(或Spectrum或…)将数据带入集群，而不是通过INSERT。这将在Redshift中产生管理事务的问题，并为VACUUM造成巨大的磁盘空间浪费/需求。将数据放入Redshift的INSERT方法是一种反模式，即使是中等大小的数据也不应该这样做。
更普遍的关注是数据移动阻抗不匹配。kineesis是由大量独立的数据流和代码生成的小文件。Redshift是一个处理大数据段的大型数据库。如果不匹配这些工具，就会错过它们的设计目标，这将使它们中的任何一个都表现得非常糟糕。您需要通过将S3批处理到Redshift来匹配数据需求。这意味着在一个COPY命令中将许多S3文件复制到Redshift中。这可以通过清单或"目录"来完成。结构。从S3路径复制所有内容…这个将数据拷贝到Redshift的过程可以每隔一段时间(2分钟、5分钟或10分钟)运行一次。因此，您希望Kinesis lambda在S3中组织数据(或添加到清单中)，以便"批处理";可以为COPY执行收集大量S3文件。通过这种方式，可以一次将大量S3文件带入Redshift(它的首选数据大小)，并且还将大大减少执行API调用。
如果你设置了一个非常大的kineesis管道并且数据非常大那么就会有另一个数据移动"首选项"考虑到这只有在每分钟移动大量数据时才有意义。这个额外的首选项是针对S3的。S3是一个对象存储，这意味着要花费大量的时间来"查找"。请求的对象键。大约是0.5秒。因此读取1000个S3对象将花费500秒(总共)500秒的键查找时间。Redshift将并行地向S3发出请求，集群中的每个片一个请求，因此其中一些时间是并行的。如果正在读取的文件大小为1KB，那么在S3查找完成后，数据的传输总共大约需要1.25秒。同样，这次是并行的，但是您可以看到在查找和传输上花费了多少时间。为了从S3获得读取许多文件所需的最大带宽，这些文件的大小需要为1GB(根据我的经验，100MB就可以了)。您可以看到，如果您每分钟要从Kinesis摄取数百万个文件到Redshift，您将需要一个进程将许多小文件合并成更大的文件，以避免S3的这种危险。由于您使用Lambda作为您的Kinesis阅读器，我希望您还没有达到这个数据速率，但如果您希望扩展到非常大的规模，那么最好关注一下这个问题。
仅仅因为工具有高带宽并不意味着它们可以连接在一起。带宽有多种形式