无需使用S3的AWS Kinesis到AWS Redshift Lambda函数开发咨询

完全可以实现无需中间S3存储桶的加载流程，核心通过Lambda直接对接Redshift接口完成流式数据写入

实现前提 #

• 确保Lambda与Redshift网络连通：Lambda如果部署在VPC内，需要配置安全组允许访问Redshift的端口；如果Redshift是公网访问模式，Lambda需配置公网出口
• Lambda执行角色已配置对应权限：redshift-data:ExecuteStatement、redshift-serverless:GetCredentials（Redshift Serverless场景）或redshift:GetClusterCredentials（预置Redshift集群场景）

两种对接场景的配置方式 #

• 对接Kinesis Data Streams：直接给Lambda配置Kinesis流作为触发器，按需设置批量触发的窗口大小和批次条数
• 对接Kinesis Firehose：开启Firehose的数据转换功能，指定自定义Lambda作为转换处理器，在Lambda处理逻辑中完成Redshift写入后按Firehose要求返回处理状态即可

核心写入逻辑示例（Python） #

推荐使用Redshift Data API实现写入，无需在Lambda中打包JDBC/ODBC驱动，依赖轻量无额外编译成本：

import boto3
import json
import base64

# 初始化Redshift Data API客户端
redshift_data = boto3.client('redshift-data')

# 替换为你自己的Redshift配置
REDSHIFT_MODE = "serverless" # 预置集群模式改为"provisioned"
REDSHIFT_WORKGROUP = "your-serverless-workgroup" # 预置集群模式改为对应CLUSTER_ID
DATABASE = "your-db-name"
TARGET_TABLE = "your-target-table"

def lambda_handler(event, context):
    batch_rows = []
    # 解析Kinesis流数据
    for record in event['Records']:
        # 解码base64编码的流数据
        raw_data = base64.b64decode(record['kinesis']['data']).decode('utf-8')
        data = json.loads(raw_data)
        # 按目标表字段顺序组装写入值，注意特殊字符转义
        batch_rows.append(f"('{data['user_id']}', '{data['event_time']}', {data['event_value']})")
    
    # 拼接批量插入SQL
    insert_sql = f"INSERT INTO {TARGET_TABLE} VALUES {','.join(batch_rows)}"
    
    # 执行写入
    if REDSHIFT_MODE == "serverless":
        resp = redshift_data.execute_statement(
            WorkgroupName=REDSHIFT_WORKGROUP,
            Database=DATABASE,
            Sql=insert_sql
        )
    else:
        resp = redshift_data.execute_statement(
            ClusterIdentifier=REDSHIFT_WORKGROUP,
            Database=DATABASE,
            Sql=insert_sql
        )
    
    # 对接Firehose场景需要返回指定格式的处理结果，可参考Firehose转换Lambda的返回格式规范调整
    return {
        "statusCode": 200,
        "executionId": resp["Id"]
    }

注意事项 #

• 控制单次写入批次大小，建议单次批量插入的行数不超过1000条，避免SQL语句过长导致执行失败
• 配置Lambda死信队列，捕获写入失败的数据避免丢失，可后续针对失败数据做重试或异常分析
• 写入前做好数据格式校验、类型转换和特殊字符转义，避免非法数据导致SQL执行报错
• 数据量极大的场景可改用COPY命令结合标准输入传递批量数据，写入性能远高于批量INSERT

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