如何将列表中的datetime日期值转换为CloudWatch Log Insight可识别的ISO格式

直接使用Python datetime内置的isoformat()方法即可完成转换，该方法输出的ISO 8601格式完全兼容CloudWatch Log Insight的日期识别规则。

完整处理示例代码 #

import datetime
from dateutil.tz import tzlocal
import json

# 原始事件列表
raw_events = [
    {
        'timestamp': datetime.datetime(2021, 9, 1, 8, 30, 25, 430000, tzinfo=tzlocal()), 
        'type': 'ExecutionFailed', 
        'id': 22, 
        'previousEventId': 21, 
        'executionFailedEventDetails': {
            'error': 'States.TaskFailed', 
            'cause': '{"AllocatedCapacity":2,"Attempt":0,"CompletedOn":1630485024021,"ErrorMessage":"An error occurred while calling o80.getDynamicFrame. The TCP/IP connection to the host 172.31.17.102, port 1433 has failed. Error: \"Connection timed out: no further information. Verify the connection properties. Make sure that an instance of SQL Server is running on the host and accepting TCP/IP connections at the port. Make sure that TCP connections to the port are not blocked by a firewall.\".","ExecutionTime":100,"GlueVersion":"2.0","Id":"jr_504a3acc2cdb7e3e18d2d22d8df69747744030e24afbcbee7f02de1ece6599c3","JobName":"sqlserver-ingest","JobRunState":"FAILED","LastModifiedOn":1630485024021,"LogGroupName":"/aws-glue/jobs","MaxCapacity":2.0,"NumberOfWorkers":2,"PredecessorRuns":[],"StartedOn":1630484906355,"Timeout":2880,"WorkerType":"G.1X"}'
        }
    }
]

# 方法1：提前遍历转换timestamp字段
processed_events = []
for event in raw_events:
    processed_event = event.copy()
    # 转换为ISO格式
    processed_event['timestamp'] = processed_event['timestamp'].isoformat()
    processed_events.append(processed_event)

# 方法2：如果要直接转JSON输出到CloudWatch，可自定义序列化函数避免提前遍历
def json_serial(obj):
    if isinstance(obj, datetime.datetime):
        return obj.isoformat()
    raise TypeError("Type %s not serializable" % type(obj))

# 直接序列化为兼容CloudWatch的JSON格式
json_output = json.dumps(raw_events, default=json_serial, ensure_ascii=False)

转换说明 #

• 转换后的timestamp格式示例为2021-09-01T08:30:25.430000+08:00，时区会根据tzlocal的实际配置自动匹配，该格式可以被CloudWatch Log Insight直接识别为日期类型，支持时间范围过滤、时间排序等操作。
• 若需要统一使用UTC时区输出，可将转换逻辑修改为processed_event['timestamp'] = processed_event['timestamp'].astimezone(datetime.timezone.utc).isoformat()。

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