如何将Amazon Kinesis KCL的checkpoint/lease表存储到本地DynamoDB

完全可以在不修改KCLPy源码的前提下实现该需求，KCLPy内部使用boto3操作DynamoDB，默认遵循boto3的配置链规则，你可以通过以下两种方式配置：

方式一：通过环境变量配置（无需修改KCL初始化代码） #

在启动KCL消费者的代码最开头设置如下环境变量即可，配置会自动被boto3读取生效：

import os

# 替换为你本地DynamoDB的实际访问地址
os.environ['AWS_DYNAMODB_ENDPOINT'] = 'http://localhost:8000'
# 本地DynamoDB默认无需鉴权，填写任意符合格式的密钥字符串即可
os.environ['AWS_ACCESS_KEY_ID'] = 'test'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'test'
# 填写任意合法的AWS区域即可
os.environ['AWS_DEFAULT_REGION'] = 'cn-north-1'

# 后续保持你原本的KCL初始化代码不变

方式二：显式传入本地DynamoDB客户端（更灵活，适合多环境切换场景） #

KCLPy的KCLProcess构造方法支持传入自定义的dynamodb_client参数，你可以手动创建指向本地实例的boto3客户端传入即可：

import boto3
from amazon_kclpy import kcl

# 创建本地DynamoDB客户端
local_dynamodb_client = boto3.client(
    'dynamodb',
    endpoint_url='http://localhost:8000',
    aws_access_key_id='test',
    aws_secret_access_key='test',
    region_name='cn-north-1'
)

# 初始化KCL时传入该客户端，其余原有配置保持不变
kcl_consumer = kcl.KCLProcess(
    record_processor_class=你的自定义记录处理类,
    dynamodb_client=local_dynamodb_client,
    application_name="你的KCL应用名",
    stream_name="你的Kinesis流名",
    region_name="cn-north-1"
)

kcl_consumer.run()

注意事项 #

• 配置前请确保本地DynamoDB服务正常启动，对应端口可正常访问
• 配置生效后KCL会自动在本地DynamoDB实例中创建lease表、存储checkpoint数据，无需额外操作
• 若你的本地DynamoDB开启了自定义鉴权，将上述示例中的测试密钥替换为实际的有效密钥即可

内容的提问来源于stack exchange，提问作者Elo Siema