AWS Lambda拆分嵌套NDJSON大文件遇TypeError问题求助

我有一个包含数千条嵌套记录的NDJSON文件，体积约150MB，希望在AWS Lambda中将其拆分为两个文件（无需严格均分），但尝试拆分时出现错误：TypeError: unhashable type: 'slice'。

NDJSON文件样本 #

{
  "time": "2023-02-28T23:28:01.295231Z",
  "resourceId": "asdf",
  "operationName": "Sign-in activity",
  "operationVersion": "1.0",
  "category": "Category",
  "resultDescription": null,
  "timestampZulu": "2023-03-08T00:05:31.495000Z",
  "timestampSeconds": 1678233931,
  "properties": {
    "alternateSignInName": null,
    "appServicePrincipalId": null,
    "appliedConditionalAccessPolicies": [
      {
        "id": "123",
        "displayName": "Guest User Access",
        "enforcedGrantControls": [
          "Guest Access Terms of Use"
        ],
        "enforcedSessionControls": [
          "SignInFrequency"
        ],
        "result": "notApplied",
        "conditionsSatisfied": 1,
        "conditionsNotSatisfied": 2
      },
      {
        "id": "123",
        "displayName": "Restricted",
        "enforcedGrantControls": [
          "Block"
        ],
        "enforcedSessionControls": [],
        "result": "notApplied",
        "conditionsSatisfied": 0,
        "conditionsNotSatisfied": 1
      }
    ],
    "authenticationContextClassReferences": [],
    "authenticationDetails": [],
    "authenticationProcessingDetails": [
      {
        "key": "adsf",
        "value": "False"
      },
      {
        "key": "ABC",
        "value": "False"
      }
    ],
    "authenticationProtocol": "none",
    "authenticationRequirement": "singleFactorAuthentication",
    "authenticationRequirementPolicies": [],
    "authenticationStrengths": [],
    "autonomousSystemNumber": 20115,
    "crossTenantAccessType": "none",
    "deviceDetail": {
      "isCompliant": false,
      "isManaged": false,
      "operatingSystem": "Windows 10"
    }
  }
}

（更多类似记录）

原Python脚本 #

def lambda_handler(event, context):
    source_key = event['Records'][0]['s3']['object']['key']
    source_bucket = event['Records'][0]['s3']['bucket']['name']
   
    # 从S3读取JSON文件内容
    response = s3.get_object(Bucket=source_bucket, Key=source_key)
    #contents = response['Body'].read().decode('utf-8')
    contents = response['Body']
    data = (line.decode('utf-8') for line in response['Body'].iter_lines())
    for row in data:
        json_data = json.loads(row)
        # df = pd.json_normalize(json_data)
        # flat_df = pd.json_normalize(df.to_dict(orient='records'))
        # flat_json = flat_df.to_json(orient='records')
   
    chunks = [json_data[i:1 + 2000] for i in range(0, len(json_data), 2000)]
   
    destination_prefix = 'split-files/'
   
    for i, chunk in enumerate(chunks):
        chunk_key = destination_prefix + f'split-{i}.json'
        chunk_content = json.dumps(chunk)
        s3.put_object(Bucket=destination_json_curated_bucket, Key=chunk_key, Body=chunk_content)
        
    return f'{len(chunks)} chunks created'
   
    # 将JSON数据均分
    # length = len(json_data)
    # print(f"Length = {length}")
    # half = length // 2
    # print(f"half = {half}")
    # first_half = json_data[:half]
    # second_half = json_data[half:]

原代码存在两个核心问题：

1. 变量赋值错误：循环读取每行NDJSON时，json_data被反复覆盖为单个记录的字典，最终只保留最后一条记录，而非所有记录的列表。
2. 切片使用错误：尝试对字典json_data使用切片语法（json_data[i:1+2000]），但字典是不可哈希的映射类型，不支持切片操作，直接触发TypeError。

以下是优化后的代码，采用流式分块处理，避免一次性加载全部数据，同时保持NDJSON格式：

import json
import boto3

s3 = boto3.client('s3')
# 替换为你的目标存储桶名称，或从环境变量读取
destination_json_curated_bucket = 'your-target-bucket'

def lambda_handler(event, context):
    source_key = event['Records'][0]['s3']['object']['key']
    source_bucket = event['Records'][0]['s3']['bucket']['name']
   
    # 流式读取S3中的NDJSON文件
    response = s3.get_object(Bucket=source_bucket, Key=source_key)
    # 过滤空行，避免JSON解析错误
    data = (line.decode('utf-8').strip() for line in response['Body'].iter_lines() if line.strip())
    
    records = []
    chunk_size = 2000  # 每个拆分文件包含2000条记录
    chunk_index = 0
    destination_prefix = 'split-files/'
    
    for row in data:
        try:
            records.append(json.loads(row))
            # 达到分块大小就写入S3
            if len(records) >= chunk_size:
                chunk_key = f"{destination_prefix}split-{chunk_index}.ndjson"
                # 转换回NDJSON格式（每行一个JSON对象）
                chunk_content = '\n'.join(json.dumps(r) for r in records)
                s3.put_object(Bucket=destination_json_curated_bucket, Key=chunk_key, Body=chunk_content)
                records = []
                chunk_index += 1
        except json.JSONDecodeError:
            # 跳过无效的JSON行，避免任务中断
            continue
    
    # 处理剩余不足chunk_size的记录
    if records:
        chunk_key = f"{destination_prefix}split-{chunk_index}.ndjson"
        chunk_content = '\n'.join(json.dumps(r) for r in records)
        s3.put_object(Bucket=destination_json_curated_bucket, Key=chunk_key, Body=chunk_content)
    
    return f"{chunk_index + 1} chunks created"

• 流式分块：边读取边收集记录，达到指定数量就写入S3，避免一次性加载150MB数据占用过多Lambda内存。
• 格式保持：拆分后的文件仍保持NDJSON格式，兼容原文件的读取逻辑。
• 错误容错：跳过无效的JSON行，避免单条记录格式错误导致整个任务失败。
• 内存高效：每次写入后清空当前记录列表，及时释放内存。

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