Python 编写 AVRO timestamp-millis: datum.astimezone(tz=timezones.utc) AttributeError: 'int' object has no attribute 'astimezone'答案

【问题标题】：Python writing AVRO timestamp-millis: datum.astimezone(tz=timezones.utc) AttributeError: 'int' object has no attribute 'astimezone'Python 编写 AVRO timestamp-millis: datum.astimezone(tz=timezones.utc) AttributeError: 'int' object has no attribute 'astimezone'
【发布时间】：2021-09-06 08:37:00
【问题描述】：

环境：Windows 10下的python 3.8.4。

我正在尝试从字典中编写 avro。

字典包含时间戳

def get_dict(self):
        return {"msg_header": {...
                                 "msg_timestamp": int(datetime.datetime.timestamp(datetime.datetime.now())),
                               ...},
                ...
                 }

Avro 架构：

{
        "name" : "msg_timestamp",
        "type" : {
          "type" : "long",
          "logicalType" : "timestamp-millis"
}

编写 Avro：

writer = avro.io.DatumWriter(schema)
bytes_writer = io.BytesIO()
encoder = avro.io.BinaryEncoder(bytes_writer)
writer.write(get_dict(), encoder) #----------Exception here
raw_bytes = bytes_writer.getvalue()

例外：

...
line
581, in write_timestamp_millis_long
    datum = datum.astimezone(tz=timezones.utc)  
AttributeError: 'int' object has no attribute 'astimezone'

任何想法如何解决它？我不想转换为 UTC。即使我将系统 TZ 设置为 UTC，它也无济于事，方法 astimezone(tz=timezones.utc) 正在执行导致异常

【问题讨论】：

datum 似乎是一个整数（Unix 时间以秒为单位），因此您必须先转换为 datetime 对象才能使用 astimezone 方法。另外，“我不想转换为 UTC”是什么意思 - 你需要哪个时区？当地时间？ Unix 时间指的是 UTC，因此您不会绕过转换。
@MrFuppes 我没有明确调用 .asdatetime，它是异常消息，它正在内部调用。我的代码在问题中提供。如果我使用 datetime.datetime.now() 而不是 int(datetime.datetime.timestamp(datetime.datetime.now())) 我得到“数据不是模式的示例”有什么方法可以告诉 python时区是UTC（其实不是UTC）避免转换？

标签： python python-3.x timestamp avro

【解决方案1】：

msg_timestamp 被定义为逻辑类型，这意味着您的字典应该包含 datetime，并且库会在序列化时自动将其转换为 long，并在反序列化时自动转换回 datetime。

所以而不是：

"msg_timestamp": int(datetime.datetime.timestamp(datetime.datetime.now()))

你只想做：

"msg_timestamp": datetime.datetime.now(tz=datetime.timezone.utc)

如果您想自己将其转换为 int 并且不让库执行此操作，请不要使用 logicalType。

【讨论】：

这会导致“数据不是架构示例”异常
抱歉，它需要时区感知，例如这样的：{"msg_timestamp": datetime.datetime.now(tz=datetime.timezone.utc)}

【解决方案2】：

最后我最终使用了 fastavro，它提供了更详细的异常消息，而不是“数据......不是模式的示例......”，它可以正常工作并处理消息列表。

import fastavro
...
parsed_schema = fastavro.schema.load_schema(schema_path)
...
def get_dict(self):
    return {"msg_header": {...
                             "msg_timestamp": int(datetime.datetime.timestamp(datetime.datetime.now())),
                           ...},
            ...
             }
...

#in a for loop do something and Prepare list of messages 
messages.append(get_dict().copy())
...

bytes_writer = io.BytesIO()
fastavro.writer(bytes_writer, parsed_schema, messages)
raw_bytes = bytes_writer.getvalue()

【讨论】：