通过AWS Firehose向Glue托管Iceberg表插入时间戳数据失败求助

针对你遇到的Firehose无法将字符串时间戳写入Iceberg表的问题，以下是几个经过验证的解决方案：

方案1：用Firehose Lambda数据转换强制修正时间戳格式 #

Glue托管的Iceberg对timestamp格式的解析严格性高于普通Hive表，Firehose的自动转换逻辑无法正确处理2024-09-04T18:56:15.114这种带T分隔符的格式。你可以通过Firehose的数据转换功能，用Lambda提前将时间戳格式转为Iceberg兼容的yyyy-MM-dd HH:mm:ss.SSS格式。

Lambda转换函数示例（Python）：

import json
from datetime import datetime

def lambda_handler(event, context):
    output = []
    for rec in event["records"]:
        # 解析原始JSON数据
        payload = json.loads(rec["data"].decode("utf-8"))
        adf_record = payload["ADF_Record"]
        
        if "baz" in adf_record:
            # 解析带T的时间戳，转换为空格分隔的格式
            ts = datetime.fromisoformat(adf_record["baz"])
            adf_record["baz"] = ts.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]  # 保留三位毫秒
        
        # 重新编码并构造输出
        output_data = json.dumps(payload).encode("utf-8")
        output.append({
            "recordId": rec["recordId"],
            "result": "Ok",
            "data": output_data
        })
    return {"records": output}

配置Firehose时启用该Lambda作为数据转换函数，确保数据在投递到Glue前完成格式修正。

方案2：调整Iceberg表的Timestamp精度与属性 #

你之前尝试的纪元格式失败是因为使用了微秒级数值（1725476175114000），而Iceberg默认timestamp类型是毫秒精度。同时，需要给表添加属性指定Parquet的时间戳格式：

1. 修改表字段精度：
   在Athena或Glue Spark作业中执行SQL：

ALTER TABLE my_db.my_table ALTER COLUMN baz SET DATA TYPE timestamp(3)

2. 给Glue表添加以下属性（在Glue控制台的表详情->编辑属性中添加）：
   • write.format.default: parquet
   • parquet.write-timestamp-format: ISO-8601

完成后，直接传递2024-09-04T18:56:15.114格式的时间戳即可被正确识别。

方案3：直接投递Parquet格式数据绕过转换 #

如果不想依赖Firehose的转换逻辑，可以在发送数据前将JSON转为Parquet格式，确保timestamp字段用正确的Arrow类型编码：

import boto3
import pyarrow as pa
import pyarrow.parquet as pq
from io import BytesIO

# 构造符合表结构的数据
data_dict = {
    "foo": ["bar"],
    "baz": [pa.scalar("2024-09-04T18:56:15.114", type=pa.timestamp('ms'))]
}
table = pa.Table.from_pydict(data_dict)

# 转为Parquet字节流
buf = BytesIO()
pq.write_table(table, buf)
buf.seek(0)
parquet_bytes = buf.read()

# 发送到Firehose
firehose = boto3.client("firehose")
firehose.put_record(
    DeliveryStreamName="my_stream",
    Record={"Data": parquet_bytes}
)

同时需修改Firehose的输入格式为Parquet，这样Firehose会直接将数据写入Iceberg表，无需类型转换。

避坑提示 #

• Glue托管Iceberg表会自动覆盖自定义SerDe参数，所以你之前尝试的timestamp.formats配置无效，不要浪费时间在这上面。
• 纪元格式必须用毫秒级数值（1725476175114），而非微秒，否则会被错误识别为date类型。

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