如何扇出AWS kinesis流？

Answer 1

有两种方法可以实现Amazon Kinesis流的扇出：

• 使用Amazon Kinesis Analytics将记录复制到其他流
• 触发AWS Lambda函数将记录复制到另一个流

选项1：使用Amazon Kinesis Analytics进行扇出

我设法实现了这一点，如下所示：

• 创建了三个流:input、output1、output2
• 创建了两个Amazon Kinesis Analytics应用程序:copy1、copy2

Amazon Kinesis Analytics SQL应用程序如下所示：

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM"
(log VARCHAR(16));

CREATE OR REPLACE PUMP "COPY_PUMP1" AS
  INSERT INTO "DESTINATION_SQL_STREAM"
    SELECT STREAM "log" FROM "SOURCE_SQL_STREAM_001";

#!/usr/bin/env python

import json, time
from boto import kinesis

kinesis = kinesis.connect_to_region("us-west-2")
i = 0

while True:
  data={}
  data['log'] =  'Record ' + str(i)
  i += 1
  print data
  kinesis.put_record("input", json.dumps(data), "key")
  time.sleep(2)

from boto import kinesis

kinesis = kinesis.connect_to_region("us-west-2")
iterator = kinesis.get_shard_iterator('output1', 'shardId-000000000000', 'TRIM_HORIZON')['ShardIterator']
records = kinesis.get_records(iterator, 5)
print [r['Data'] for r in records['Records']]

[u'{"LOG":"Record 0"}', u'{"LOG":"Record 1"}', u'{"LOG":"Record 2"}', u'{"LOG":"Record 3"}', u'{"LOG":"Record 4"}']

选项2：使用AWS Lambda

如果要向多个流展开，一个更有效的方法可能是创建一个AWS Lambda函数：

• 由Amazon Kinesis流记录触发
• 将记录写入多个Amazon Kinesis“输出”流