Der Konsum einer kinesis-stream in python

sollten Sie verwenden, boto.kinesis:

from boto import kinesis

Nachdem Sie erstellt einen stream:

Schritt 1: anschließen an die aws kinesis:

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

Schritt 2: Holen Sie sich die stream-info (wie, wie viele Scherben, wenn es aktiv ist ..)

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...')

Schritt 3 : Holen Sie sich alle Splitter ids, und für jede freigegebene id-Holen Sie sich die shard iterator:

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'] })

Schritt 4 : Lesen Sie die Daten für die einzelnen Splitter

limit ist das limit der Datensätze, die Sie erhalten möchten. (Sie können bis zu 10 MB)
shard_iterator ist das gemeinsame aus den vorherigen Schritt.

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

in Ihrem nächsten Aufruf get_records, sollten Sie die shard_iterator, die Sie erhalten mit dem Ergebnis der vorherigen get_records.

Hinweis: in einem Aufruf zu get_records, (limit = None) erhalten Sie leere Datensätze.
wenn der Aufruf zu get_records mit einem limit, erhalten Sie die Datensätze in der gleichen partition key (wenn Sie Daten zu streamen, musst du den partition key :

connection.put_record(stream_name, data, partition_key)