如何通过Kinesis Transform Lambda将任意格式CSV日志转为JSON

要实现任意CSV格式转JSON，核心是把字段映射和分隔符做成可配置项，避免硬编码。下面是具体实现思路和Python脚本：

核心思路 #

• 用Lambda环境变量存储CSV的字段名列表和分隔符，比如FIELD_NAMES设为event,recordtime,remotehost,...，DELIMITER设为|
• 动态解析每条CSV记录，根据配置的字段名生成对应JSON
• 适配Kinesis Firehose的批量处理逻辑，处理base64编码的输入记录

完整Python Lambda脚本 #

import base64
import json
import os

def lambda_handler(event, context):
    # 从环境变量读取配置
    field_names = os.environ.get('FIELD_NAMES', '').split(',')
    delimiter = os.environ.get('DELIMITER', '|')
    # 处理Firehose的批量记录
    output_records = []
    
    for record in event['records']:
        # 解码base64的日志内容
        payload = base64.b64decode(record['data']).decode('utf-8').strip()
        # 拆分CSV字段（处理可能的空值）
        csv_fields = [field.strip() for field in payload.split(delimiter)]
        
        # 生成JSON：字段名和CSV字段一一对应，长度不足的补空字符串
        json_payload = {}
        for idx, field_name in enumerate(field_names):
            if idx < len(csv_fields):
                json_payload[field_name] = csv_fields[idx]
            else:
                json_payload[field_name] = ''
        
        # 转成JSON字符串并重新编码为base64（加换行符适配后续Firehose处理）
        json_str = json.dumps(json_payload) + '\n'
        encoded_payload = base64.b64encode(json_str.encode('utf-8')).decode('utf-8')
        
        # 构造符合Firehose要求的输出记录
        output_records.append({
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': encoded_payload
        })
    
    return {'records': output_records}

配置与扩展说明 #

1. Lambda环境变量设置：
   • FIELD_NAMES：逗号分隔的字段名，比如针对示例日志设为event,recordtime,remotehost,remoteport,pid,dbname,username,authmethod,duration,sessionid
   • DELIMITER：CSV的分隔符，示例里是|，标准CSV可设为,
2. 处理带引号的复杂CSV：
   如果遇到带引号包裹字段的CSV（比如字段内容含分隔符），可以替换拆分逻辑为csv.reader处理：

   import csv
   from io import StringIO
   # 替换原有拆分csv_fields的代码
   reader = csv.reader(StringIO(payload), delimiter=delimiter)
   csv_fields = next(reader)
   

3. 适配不同格式场景：
   切换其他CSV格式时，仅需更新FIELD_NAMES和DELIMITER的环境变量值，无需修改代码。若列数多于字段名，多余列会被忽略；列数少于字段名时，缺失字段会设为空字符串。

注意事项 #

• 确保日志内容为UTF-8编码，避免解码错误
• 严格遵循Firehose的返回格式要求，每条记录必须包含recordId、result和base64编码的data

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