我正在將 Boto3 與 Kinesis 一起使用。我希望 Kinesis 從流中回傳單個專案。
這是我的代碼：
kinesis = boto3.client('kinesis')
iterator = kinesis.get_shard_iterator(
StreamName = "requests",
ShardIteratorType='LATEST'
)
response = kinesis.get_records(
ShardIterator = iterator,
Limit = 1,
StreamName = "requests",
)
print(response)
它失敗并出現此錯誤：
botocore.exceptions.ParamValidationError: Parameter validation failed:
Missing required parameter in input: "ShardId"
在我初始化迭代器的那一行，所以我知道我缺少一個引數，但是我要設定ShardId什么？如果答案因流而異，那么 a 到底是什么ShardId？
提前致謝！
uj5u.com熱心網友回復：
核心問題是get_shard_iterator需要ShardId傳遞給它的引數。
分片 ID 是流中分片的唯一識別符號。
要為您的流獲取ShardId（或ShardId從中選擇），首先呼叫 describe_stream.
stream = client.describe_stream(StreamName='requests')
這將向您回傳一個回應物件，其中包含一個 JsonPath 為response.StreamDescription.Shards.
此陣列中的每個物件都有一個ShardId. 根據您配置 Kinesis 流的方式，您將在此陣列中擁有 1 個或多個分片物件。
如果您有 1 個分片，下面應該允許您選擇第一個分片 ID 并繼續執行其余代碼：
kinesis = boto3.client('kinesis')
stream = client.describe_stream(StreamName='requests')
shard_id = stream['StreamDescription']['Shards'][0]['ShardId']
iterator = kinesis.get_shard_iterator(
StreamName = "requests",
ShardIteratorType='LATEST',
ShardId = shard_id
)
response = kinesis.get_records(
ShardIterator = iterator,
Limit = 1
)
print(response)
要了解有關分片本身的更多資訊（超出范圍），請先閱讀John Rotenstein的這個很棒的SO 答案。
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/457750.html