如何使用Python Boto3客户端正确读取多分片Kinesis流并做检查点？

以下是针对长期运行进程读取Kinesis流的完整实现方案，涵盖分片发现、检查点管理、重新分片处理等核心步骤：

核心思路 #

Lambda读取Kinesis时，AWS自动处理分片分配、检查点和重新分片；自定义长期进程则需要手动实现这些逻辑：定期发现分片、跟踪每个分片的处理进度、处理分片的开闭状态、批量记录检查点。

实现步骤与代码示例 #

1. 依赖与初始化 #

确保安装boto3和botocore（用于异常重试逻辑）：

pip install boto3 botocore

初始化Kinesis客户端：

import boto3
from botocore.exceptions import ClientError, ProvisionedThroughputExceededException
import time
from typing import Dict, Optional

kinesis_client = boto3.client('kinesis', region_name='你的区域')
STREAM_NAME = '你的Kinesis流名称'

2. 检查点管理 #

检查点用于记录每个分片的最后处理位置，这里以本地文件存储为例（分布式部署建议改用DynamoDB）：

class CheckpointManager:
    def __init__(self, checkpoint_file: str = 'kinesis_checkpoints.json'):
        self.checkpoint_file = checkpoint_file
        self.checkpoints = self._load_checkpoints()

    def _load_checkpoints(self) -> Dict[str, str]:
        try:
            with open(self.checkpoint_file, 'r') as f:
                import json
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return {}

    def save_checkpoint(self, shard_id: str, sequence_number: str):
        self.checkpoints[shard_id] = sequence_number
        with open(self.checkpoint_file, 'w') as f:
            import json
            json.dump(self.checkpoints, f)

    def get_checkpoint(self, shard_id: str) -> Optional[str]:
        return self.checkpoints.get(shard_id)

3. 分片发现 #

定期调用list_shards()获取当前所有活跃分片，过滤已处理完成的关闭分片：

def get_active_shards() -> list:
    try:
        response = kinesis_client.list_shards(StreamName=STREAM_NAME)
        # 只保留ACTIVE状态的分片，CLOSED分片处理完记录后即可放弃
        return [shard for shard in response['Shards'] if shard['ShardStatus'] == 'ACTIVE']
    except ClientError as e:
        print(f"获取分片失败: {e}")
        return []

4. 单分片处理逻辑 #

处理单个分片的循环读取、记录检查点：

def process_shard(shard_id: str, checkpoint_manager: CheckpointManager):
    # 获取分片迭代器：有检查点则从指定位置开始，否则从流的起始位置读取
    checkpoint = checkpoint_manager.get_checkpoint(shard_id)
    if checkpoint:
        iterator_response = kinesis_client.get_shard_iterator(
            StreamName=STREAM_NAME,
            ShardId=shard_id,
            ShardIteratorType='AFTER_SEQUENCE_NUMBER',
            StartingSequenceNumber=checkpoint
        )
    else:
        iterator_response = kinesis_client.get_shard_iterator(
            StreamName=STREAM_NAME,
            ShardId=shard_id,
            ShardIteratorType='TRIM_HORIZON'
        )
    
    shard_iterator = iterator_response['ShardIterator']
    batch_count = 0  # 批量计数，每处理N批记录后保存检查点

    while shard_iterator:
        try:
            records_response = kinesis_client.get_records(ShardIterator=shard_iterator, Limit=10000)
            records = records_response['Records']
            
            if records:
                # 替换为你的业务处理逻辑，这里示例打印记录内容
                for record in records:
                    data = record['Data'].decode('utf-8')
                    print(f"分片 {shard_id} 处理记录: {data}")
                
                # 每处理10批记录保存一次检查点，避免频繁IO
                batch_count += 1
                if batch_count >= 10:
                    last_sequence = records[-1]['SequenceNumber']
                    checkpoint_manager.save_checkpoint(shard_id, last_sequence)
                    batch_count = 0
            
            # 更新分片迭代器，若为None说明分片已关闭
            shard_iterator = records_response.get('NextShardIterator')
            
            # 控制读取频率，避免过度调用API
            time.sleep(0.1)

        except ProvisionedThroughputExceededException:
            # 遇到限流时，指数退避重试
            time.sleep(2)
        except ClientError as e:
            print(f"读取分片 {shard_id} 失败: {e}")
            time.sleep(1)

5. 主循环：全局分片管理 #

定期刷新分片列表，启动每个分片的处理（单进程用线程实现多分片并发处理）：

import threading

def main():
    checkpoint_manager = CheckpointManager()
    active_shards = set()  # 跟踪当前正在处理的分片ID

    while True:
        # 每30秒刷新一次分片列表，处理重新分片情况
        current_shards = get_active_shards()
        current_shard_ids = {shard['ShardId'] for shard in current_shards}

        # 启动新分片的处理线程
        for shard in current_shards:
            shard_id = shard['ShardId']
            if shard_id not in active_shards:
                thread = threading.Thread(target=process_shard, args=(shard_id, checkpoint_manager))
                thread.daemon = True
                thread.start()
                active_shards.add(shard_id)
        
        # 移除已关闭的分片（分片状态变为CLOSED后，线程会自动退出）
        active_shards.intersection_update(current_shard_ids)
        
        time.sleep(30)

if __name__ == "__main__":
    main()

关键注意事项 #

• 重新分片处理：主循环每30秒刷新分片列表，自动发现新拆分/合并的分片；旧的CLOSED分片处理完所有记录后，线程会因NextShardIterator为None而自动退出。
• 检查点策略：避免每条记录都保存检查点，建议按批量或时间间隔保存，平衡数据一致性与性能。
• 限流与重试：处理ProvisionedThroughputExceededException时使用指数退避，避免被Kinesis限流。
• 分片迭代器过期：分片迭代器默认15分钟过期，若进程暂停超过15分钟，重启时会通过检查点的序列号重新获取迭代器，不会丢失处理进度。

内容的提问来源于stack exchange，提问作者Ivan Voras