Потребитель Kinesis возвращает пустую запись (boto, python) - PullRequest
Новая
       18

Потребитель Kinesis возвращает пустую запись (boto, python)

0 голосов
/ 09 мая 2018

У меня проблемы с проверкой данных, которые я пишу в Kinesis. Кажется, что следующий пример должен работать, но я получаю пустой список, возвращаемый из get_records (в ​​поле Records). Есть идеи, что может происходить? 

import uuid
import boto3
import time


streamname = 'mytestStream'
session = boto3.session.Session() 
kinesis_client = session.client('kinesis', region_name='us-east-1')


##### WRITE TO KINESIS

partitionkey = str(uuid.uuid4())[:8]
put_response = kinesis_client.put_record(StreamName=streamname,Data='mytestdata',PartitionKey=partitionkey)

time.sleep(5)


##### READ FROM KINESIS

shard_id = kinesis_client.describe_stream(StreamName=streamname)['StreamDescription']['Shards'][0]['ShardId']
shard_iterator = kinesis_client.get_shard_iterator(StreamName=streamname, ShardId=shard_id, ShardIteratorType="LATEST")["ShardIterator"]
data_from_kinesis = kinesis_client.get_records(ShardIterator=shard_iterator)

Спасибо! * * 1004

1 Ответ

0 голосов
/ 11 мая 2018

Если вы будете использовать контрольную точку LATEST, вы должны сначала начать чтение потока, а затем поместить запись. В вашем примере временная шкала выглядит следующим образом:

  • в t0: последняя контрольная точка в потоке находится в 101.
  • в t1 (основной поток): запись помещается в поток, а запись находится на контрольной точке 102.
  • в момент времени t2 (основной поток): вы начинаете «хвост» потока с ПОСЛЕДНЕЙ точки, равной 103.

Чтобы это исправить, вы должны запустить производителя и потребителя в разных потоках. Правильный поток должен быть таким:

  • в t0 (потребительская нить): начните отводить пар в ПОСЛЕДНЮЮ позицию, которая равна 201.
  • в t1 (поток производителя): вы помещаете запись в поток, и запись помещается в контрольную точку 202.
  • в момент t2 (потребительский поток): когда осколок на стороне сервера движется вперед (потому что вы только что добавили данные), и вы были в хвосте со времени контрольной точки 201, вы повторяете новую контрольную точку 202, и ваши данные отображаются.
...