AWS Kinesis Firehose动态分区：日期解析失败问题求助

问题背景 #

接收的JSON payload格式如下：

{
  "clientId": "ASGr496mndGs80oCC97mf",
  "createdAt": "2022-09-21T14:44:53.708Z",
  ...
}

已启用Firehose的「动态分区」和「内联JSON解析」，配置的分区键如下：

year 
.createdAt| strptime("%Y-%m-%dT%H:%M:%S.%fZ")| strftime("%Y")
month 
.createdAt| strptime("%Y-%m-%dT%H:%M:%S.%fZ")| strftime("%m")
day
.createdAt| strptime("%Y-%m-%dT%H:%M:%S.%fZ")| strftime("%d")
hour
.createdAt| strptime("%Y-%m-%dT%H:%M:%S.%fZ")| strftime("%h")

报错信息：date "2022-09-21T18:30:04.431Z" does not match format "%Y-%m-%dT%H:%M:%S.%fZ"，原因是strptime的%f要求6位微秒，但数据中只有3位毫秒。

解决方案 #

方案1：修复日期解析格式 #

通过字符串替换把3位毫秒补全为6位，让strptime可以正常解析：

year 
.createdAt | sub("\\.(\\d{3})Z$"; ".\\1000Z") | strptime("%Y-%m-%dT%H:%M:%S.%fZ") | strftime("%Y")
month 
.createdAt | sub("\\.(\\d{3})Z$"; ".\\1000Z") | strptime("%Y-%m-%dT%H:%M:%S.%fZ") | strftime("%m")
day
.createdAt | sub("\\.(\\d{3})Z$"; ".\\1000Z") | strptime("%Y-%m-%dT%H:%M:%S.%fZ") | strftime("%d")
hour
.createdAt | sub("\\.(\\d{3})Z$"; ".\\1000Z") | strptime("%Y-%m-%dT%H:%M:%S.%fZ") | strftime("%H")

注意：小时的格式化要用%H（24小时制），原配置中的%h是月份缩写，会导致错误。

方案2：直接截取字符串（推荐，无需日期解析） #

因为仅关注年、月、日、小时，可直接从createdAt字符串中截取对应部分，完全避免日期解析问题：

year 
.createdAt | .[0:4]
month 
.createdAt | .[5:7]
day
.createdAt | .[8:10]
hour
.createdAt | .[11:13]

这种方式更高效，也不会出现格式匹配问题。

方案3：使用Lambda预处理数据 #

如果需要更复杂的日期处理，可在Firehose数据流前添加Lambda函数，将createdAt转换为符合%f格式的字符串，或者直接在JSON中新增year、month、day、hour字段，后续Firehose直接使用这些字段作为分区键。

示例Lambda处理逻辑（Python）：

import json
import base64

def lambda_handler(event, context):
    output = []
    for record in event['records']:
        payload = json.loads(base64.b64decode(record['data']))
        # 提取日期部分
        dt_str = payload['createdAt']
        payload['year'] = dt_str[:4]
        payload['month'] = dt_str[5:7]
        payload['day'] = dt_str[8:10]
        payload['hour'] = dt_str[11:13]
        # 转换回base64
        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(json.dumps(payload).encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)
    return {'records': output}

之后在Firehose中配置分区键为.year、.month、.day、.hour即可。

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