Блог

У меня есть следующий блок Python, чтобы получить записи, которые я поместил в поток Kinesis, а затем поместить записи в корзину S3. Для этого потока существует один сегмент.

 # Fetching the shard iterators from the Kinesis stream
shard_iterators = []
if response and 'StreamDescription' in response:
    for shard_id in response['StreamDescription']['Shards']:
        shard_id = shard_id['ShardId']
        shard_iterator = kinesis_connection.get_shard_iterator(stream_name, shard_id, 'LATEST')
        shard_iterators.append(shard_iterator['ShardIterator'])

# Iterating over the Kinesis stream and pushing data to S3
bucket = s3_connection.get_bucket(bucket_name)
k = Key(bucket)

for shard_iterator in shard_iterators:
    while 1:
        response = kinesis_connection.get_records(shard_iterator)
        shard_iterator = response['NextShardIterator']
        if len(response['Records'])> 0:
            for res in response['Records']:
                k.key = datetime.datetime.now().strftime('%Y/%m/%d/')   res['SequenceNumber']
                k.set_contents_from_string(res['Data'])
  

Данные изначально извлекаются из потока и передаются в S3, но в какой-то момент я получаю TypeError: expected string or buffer по линии response = kinesis_connection.get_records(shard_iterator) . В сегменте есть еще много записей, которые должны быть доступны для меня. Кто-нибудь сталкивался с этой проблемой раньше и / или у кого-нибудь есть идеи о том, что я должен попытаться исправить?

Эта ошибка, вероятно, вызвана декодированием json, не являющимся строкой. например None , или int .

 import json
json.loads(None)
...
File "/usr/lib/python2.7/json/decoder.py", line 366, in decode
   obj, end = self.raw_decode(s, idx=_w(s, 0).end())
TypeError: expected string or buffer
  

Что может происходить здесь: https://github.com/boto/boto/blob/develop/boto/kinesis/layer1.py#L705

Это означало бы, что тело ответа возвращается None .

Это может быть связано с истечением срока действия вашего shard_iterator. Итераторы сегментов работают только в течение 5 минут, поэтому, если обработка текущего пакета записей занимает много времени или иным образом останавливается, это может быть так. СМОТРИТЕ: http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html

Другая возможность заключается в том, что это прерывистая ошибка авторизации / подключения, которую я видел раньше, и которая может генерировать ложные исключения, которые могут привести к тому же пути кода boto?

Хотя, если бы вы могли предоставить больше трассировки стека и любых сообщений журнала, которые вы видите, это было бы полезно.