如何格式化Timestamp以将JSON数据导入AWS IoT Analytics Parquet数据存储

问题场景 #

通道原始数据格式：

{
  "Total_in": 1825.5841,
  "Time": "2023-02-17T14:08:19"
}

目标Parquet Schema：

之前尝试的秒级/毫秒级时间戳、%Y-%m-%dT%H:%M:%S格式均导致无记录进入数据存储；使用%Y-%m-%dT%H:%M:%S..%fZ格式时，数据存储显示有消息到达，但查询无结果。

解决方案 #

方案1：使用管道内置SQL转换活动 #

AWS IoT Analytics支持Presto SQL语法，可直接通过函数将原时间字符串转为TIMESTAMP类型，同时修正字段名大小写：

SELECT 
    -- 将原ISO格式字符串转为TIMESTAMP类型
    STR_TO_TIMESTAMP(Time, '%Y-%m-%dT%H:%i:%s') AS time,
    -- 转换数值类型并重命名字段
    CAST(Total_in AS FLOAT) AS total_in
FROM input

如果需要强制带UTC时区标识（避免时区解析歧义），可修改为：

SELECT 
    CONCAT(Time, '.000Z') AS time,
    CAST(Total_in AS FLOAT) AS total_in
FROM input

方案2：使用Lambda转换活动 #

编写Python Lambda函数处理数据格式，确保时间字段符合Parquet TIMESTAMP的ISO 8601标准格式（带毫秒和时区）：

import json
import base64

def lambda_handler(event, context):
    transformed_records = []
    for record in event['records']:
        # 解码原始数据
        payload = json.loads(base64.b64decode(record['data']).decode('utf-8'))
        # 格式化时间：添加毫秒和UTC时区标识
        formatted_time = f"{payload['Time']}.000Z"
        # 构造转换后的数据（字段名需与Parquet Schema完全匹配）
        transformed_payload = {
            'time': formatted_time,
            'total_in': payload['Total_in']
        }
        # 编码返回
        transformed_data = base64.b64encode(json.dumps(transformed_payload).encode('utf-8')).decode('utf-8')
        transformed_records.append({
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': transformed_data
        })
    return {'records': transformed_records}

关键注意事项 #

1. 字段名严格匹配：Parquet Schema中的字段是小写的time和total_in，转换时必须将原字段Time、Total_in重命名为对应小写名称，否则数据无法正确映射。
2. 时间格式正确性：Parquet TIMESTAMP支持的标准格式为YYYY-MM-DDTHH:mm:ss.SSSZ（毫秒级+UTC时区），之前使用的%Y-%m-%dT%H:%M:%S..%fZ格式存在格式错误（多余的点），导致数据解析失败。
3. 排查建议：可将转换后的数据临时输出到S3，通过Parquet查看工具（如Apache Drill、Pandas）检查字段类型和数据内容，确认转换是否符合预期。

内容的提问来源于stack exchange，提问作者geri-m