从Lambda向Step Functions传递DataFrame时遇413错误的解决方案咨询

错误原因 #

Step Functions 的 StartExecution 接口对输入内容大小有严格限制（最大1MB），你的JSON数据体积超过了这个阈值，导致触发413错误。

方案1：压缩数据后传递（推荐，无需额外存储服务） #

通过gzip压缩JSON数据并转成base64字符串，大幅缩小体积后传给状态机，状态机内的处理步骤再解码还原数据。

原Lambda修改代码：

import json
import boto3
import gzip
import base64

# 构建原始数据字典
data_dict = {
    'username': arr[0],
    'region': arr[1],
    'country': arr[2],
    'grid': arr[3],
    'physicalServers': arr[4],
    'servers': arr[5]
}

# 压缩并编码数据
json_bytes = json.dumps(data_dict).encode('utf-8')
compressed_data = gzip.compress(json_bytes)
base64_encoded = base64.b64encode(compressed_data).decode('utf-8')

# 调用状态机，传递压缩后的数据
client = boto3.client('stepfunctions')
response = client.start_execution(
    stateMachineArn='arn:aws:states:us-west-2:##:stateMachine:MyStateMachineTest',
    name='testStateMachine',
    input=json.dumps({'compressed_data': base64_encoded})
)

print(response)

状态机首个Lambda处理代码：

import json
import gzip
import base64

def lambda_handler(event, context):
    # 解码并解压数据
    base64_data = event['compressed_data']
    compressed_bytes = base64.b64decode(base64_data)
    original_json = gzip.decompress(compressed_bytes).decode('utf-8')
    data_dict = json.loads(original_json)
    
    # 后续数据清理、处理逻辑
    return data_dict

注意：若压缩后仍超1MB，可将dataframe拆分为多个子部分分别压缩传递，状态机内再合并。

方案2：前置数据精简 #

如果状态机不需要处理全部字段，先在接收Lambda中过滤冗余列/行、做数据聚合，将数据体积压缩至1MB以内再传递。

示例代码：

import json
import boto3
import pandas as pd

# 加载并精简physicalServers dataframe
physical_servers_df = pd.read_json(arr[4])
# 仅保留状态机需要的字段
filtered_physical = physical_servers_df[['id', 'name', 'status']].to_json()

# 加载并精简servers dataframe
servers_df = pd.read_json(arr[5])
filtered_servers = servers_df[['server_id', 'region', 'usage']].to_json()

# 构建精简后的字典
data_dict = {
    'username': arr[0],
    'region': arr[1],
    'country': arr[2],
    'grid': arr[3],
    'physicalServers': filtered_physical,
    'servers': filtered_servers
}

# 调用状态机
client = boto3.client('stepfunctions')
response = client.start_execution(
    stateMachineArn='arn:aws:states:us-west-2:##:stateMachine:MyStateMachineTest',
    name='testStateMachine',
    input=json.dumps(data_dict)
)

print(response)

方案3：分批次调用状态机 #

将大dataframe拆分为多个小批次，原Lambda循环触发状态机处理每个批次，状态机需支持批次进度记录与最终汇总。

示例代码：

import json
import boto3
import pandas as pd

# 拆分physicalServers dataframe为100行/批次
physical_df = pd.read_json(arr[4])
batch_size = 100
physical_batches = [
    physical_df[i:i+batch_size].to_json() 
    for i in range(0, len(physical_df), batch_size)
]

client = boto3.client('stepfunctions')
# 循环触发状态机处理每个批次
for idx, batch in enumerate(physical_batches):
    batch_data = {
        'username': arr[0],
        'region': arr[1],
        'country': arr[2],
        'grid': arr[3],
        'data_type': 'physicalServers',
        'batch_index': idx,
        'total_batches': len(physical_batches),
        'batch_data': batch
    }
    response = client.start_execution(
        stateMachineArn='arn:aws:states:us-west-2:##:stateMachine:MyStateMachineTest',
        name=f'testStateMachine-physical-batch-{idx}',
        input=json.dumps(batch_data)
    )
    print(f"启动批次 {idx}：{response}")

# 同理拆分并处理servers dataframe

状态机需添加分支逻辑：处理单批次数据→记录进度→判断是否所有批次完成→执行汇总操作。

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