Python中如何生成符合Avro timestamp-micros逻辑类型的long型时间戳？

错误原因 #

你直接传入Python原生datetime对象不符合Schema要求——该字段的Schema定义是[null, {type: "long", logicalType: "timestamp-micros"}]，要求传入的是从UTC epoch（1970-01-01 00:00:00）开始计算的微秒级时间戳（long类型），或是通过Avro逻辑类型适配的带时区datetime对象。

两种可行解决方案 #

方案1：手动计算微秒级时间戳（转long类型）

直接计算UTC时间对应的微秒数，转为整数后传入：

import datetime

def get_utc_timestamp_micros():
    # 定义UTC epoch时间点
    epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
    # 获取当前UTC时间
    current_utc = datetime.datetime.now(datetime.timezone.utc)
    # 计算时间差并转为微秒数，返回整数
    return int((current_utc - epoch).total_seconds() * 1_000_000)

# 写入数据时调用函数
schema_parsed = make_avsc_object(schema)

with open('images.avro', 'wb') as f:
    writer = DataFileWriter(f, DatumWriter(), schema_parsed)
    writer.append({'image_id': 'image-1', 'image_byte': a, 'update_time': get_utc_timestamp_micros()})
    writer.append({'image_id': 'image-2', 'image_byte': a, 'update_time': get_utc_timestamp_micros()})
    writer.close()

方案2：利用Avro逻辑类型自动转换（更简洁）

如果使用avro-python3库，可通过LogicalDatumWriter自动处理带时区的datetime对象，无需手动计算时间戳：

from avro.datafile import DataFileWriter
from avro.io import LogicalDatumWriter
import datetime

schema_parsed = make_avsc_object(schema)

with open('images.avro', 'wb') as f:
    # 使用LogicalDatumWriter替代默认的DatumWriter，支持逻辑类型转换
    writer = DataFileWriter(f, LogicalDatumWriter(), schema_parsed)
    # 必须传入带UTC时区的datetime对象，不能用无时区的datetime.now()
    writer.append({'image_id': 'image-1', 'image_byte': a, 'update_time': datetime.datetime.now(datetime.timezone.utc)})
    writer.append({'image_id': 'image-2', 'image_byte': a, 'update_time': datetime.datetime.now(datetime.timezone.utc)})
    writer.close()

注意事项 #

• 必须使用UTC时间：Avro的timestamp-micros逻辑类型基于UTC标准，使用本地时间会导致时间偏差。
• 若要传入null值，直接写update_time: None即可。

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