在python中消耗运动流

Question

我似乎找不到一个像样的例子来说明如何通过Python使用AWS动态流。有人能给我举几个例子吗？

最好的

Answer 1

虽然这个问题已经得到了回答，但对于未来的读者来说，考虑使用Python的Kinesis客户端库（KCL）而不是直接使用boto可能是个好主意。当您有多个消费者实例和/或更改分片配置时，它简化了从流中的消费。

https://aws.amazon.com/blogs/aws/speak-to-kinesis-in-python/

更完整地列举KCL提供的内容

• 连接到流
• 枚举碎片
• 与其他工作人员（如果有）协调碎片关联
• 为其管理的每个碎片实例化一个记录处理器
• 从流中提取数据记录
• 将记录推送到相应的记录处理器
• 检查点处理记录（它使用DynamoDB，因此您的代码不必手动保留检查点值）
• 当工作者实例计数更改时，平衡碎片工作者关联
• 分割或合并碎片时平衡碎片工作者关联

粗体的项目是我认为KCL在boto上真正提供了非平凡价值的项目。但是，根据您的用例，boto可能要简单得多。

Answer 2

你应该使用boto。运动：

from boto import kinesis

创建流后：

第1步：连接到aws kinesis：

auth = {"aws_access_key_id":"id", "aws_secret_access_key":"key"}
connection = kinesis.connect_to_region('us-east-1',**auth)

第2步：获取流信息（例如有多少分片，如果它是活动的...）

tries = 0
while tries < 10:
    tries += 1
    time.sleep(1)
    try:
        response = connection.describe_stream('stream_name')   
        if response['StreamDescription']['StreamStatus'] == 'ACTIVE':
            break 
    except :
        logger.error('error while trying to describe kinesis stream : %s')
else:
    raise TimeoutError('Stream is still not active, aborting...')

步骤3：获取所有碎片id，并为每个共享id获取碎片迭代器：

shard_ids = []
stream_name = None 
if response and 'StreamDescription' in response:
    stream_name = response['StreamDescription']['StreamName']                   
    for shard_id in response['StreamDescription']['Shards']:
         shard_id = shard_id['ShardId']
         shard_iterator = connection.get_shard_iterator(stream_name, shard_id, shard_iterator_type)
         shard_ids.append({'shard_id' : shard_id ,'shard_iterator' : shard_iterator['ShardIterator'] })

步骤4：读取每个碎片的数据

limit是您希望接收的记录的限制。（最多可以接收10 MB）shard_迭代器是上一步的共享html" target="_blank">迭代器。

tries = 0
result = []
while tries < 100:
     tries += 1
     response = connection.get_records(shard_iterator = shard_iterator , limit = limit)
     shard_iterator = response['NextShardIterator']
     if len(response['Records'])> 0:
          for res in response['Records']: 
               result.append(res['Data'])                  
          return result , shard_iterator

在下一次打电话给get_records时，您应该使用上一次get_records结果中收到的shard_iterator。

注意：在一次获取_记录的调用中，（limit=None），您可以收到空记录。如果调用get_记录时有限制，您将获得相同分区键中的记录（当您将数据放入流时，您必须使用分区键：

connection.put_record(stream_name, data, partition_key)