能否在AWS Lambda中使用BigQuery？如何将S3/DynamoDB数据同步至BigQuery

一、Lambda与BigQuery集成可行性及实现方案 #

可以直接将BigQuery与AWS Lambda集成，是成熟的实现方案，你当前的S3 Put事件触发Lambda写入BigQuery的逻辑完全可行，具体实现步骤如下：

• 依赖配置：Python开发环境需要安装google-cloud-bigquery和google-auth依赖，打包到Lambda部署包或者制作为Lambda层使用，注意要打包适配AWS Lambda Linux运行环境的二进制版本，避免本地打包的mac/Windows版本无法在Lambda环境运行。
• 权限配置：GCP侧创建拥有BigQuery数据写入、作业创建权限的服务账号，导出JSON格式密钥，将密钥内容存储到AWS Secrets Manager；给Lambda执行角色配置S3对象读取权限、Secrets Manager密钥读取权限、Lambda基础日志权限。
• 触发配置：给目标S3桶添加事件通知，触发事件选择All object create events（可按需仅选择Put事件），触发目标绑定你开发的Lambda函数，可根据你Firehose写入的文件格式配置后缀过滤（比如.json、.parquet），减少无效触发。
• 核心逻辑示例：

import boto3
import json
from google.cloud import bigquery
from google.oauth2 import service_account

s3_client = boto3.client("s3")
sm_client = boto3.client("secretsmanager")

# 初始化BigQuery客户端（冷启动仅执行一次）
gcp_secret = sm_client.get_secret_value(SecretId="你的GCP服务账号密钥Secret名称")
gcp_cred_info = json.loads(gcp_secret["SecretString"])
credentials = service_account.Credentials.from_service_account_info(gcp_cred_info)
bq_client = bigquery.Client(credentials=credentials, project="你的GCP项目ID")
BIGQUERY_TABLE_ID = "GCP项目ID.数据集名.表名"

def lambda_handler(event, context):
    for record in event["Records"]:
        # 解析S3事件参数
        bucket = record["s3"]["bucket"]["name"]
        obj_key = record["s3"]["object"]["key"]
        # 读取S3文件内容
        file_content = s3_client.get_object(Bucket=bucket, Key=obj_key)["Body"].read().decode("utf-8")
        upload_data = json.loads(file_content)
        # 写入BigQuery
        write_errors = bq_client.insert_rows_json(BIGQUERY_TABLE_ID, upload_data)
        if write_errors:
            # 实际使用可将失败请求转发至死信队列后续补数
            raise Exception(f"BigQuery写入失败：{write_errors}")
    return {"code": 0, "msg": "写入完成"}

二、DynamoDB流式同步到BigQuery的其他Python方案 #

除了你当前使用的Dynamo Stream + Kinesis Firehose落S3再同步的方案外，还有以下适配Python开发的方案可选：

• 方案1：Dynamo Stream直接触发Lambda同步。给DynamoDB开启流，配置流事件触发Python Lambda函数，函数内解析流中的INSERT/UPDATE/DELETE事件，转换为目标格式后直接写入BigQuery，该方案延迟最低可到秒级，适合低延迟同步需求。
• 方案2：Glue流式ETL作业同步。创建AWS Glue流式作业，选择Python作为开发语言，作业源对接DynamoDB流，在作业内完成数据清洗、格式转换等逻辑后，通过内置的BigQuery连接器直接写入目标表，适合有复杂数据转换需求的场景，无需自行维护运行环境。
• 方案3：Kinesis Data Streams中转同步。将Dynamo Stream数据投递到Kinesis Data Streams，使用Python版Kinesis Client Library（KCL）开发消费端应用，批量拉取流数据写入BigQuery，适合写入流量波动大的场景，可通过Kinesis做削峰填谷，避免写入突增压垮下游。

注意事项：当前S3触发Lambda的方案建议配置死信队列，将写入失败的事件投递到SQS存储，方便后续排查补数，避免数据丢失；批量写入BigQuery时建议攒够一定数据量再提交请求，降低API调用频次，减少GCP侧成本。

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