DataFrame.collect()返回Row类型时显示datetime.datetime，如何仅保留时间戳？

问题描述 #

我通过dataframe.collect()将DataFrame转换为Row类型进行后续处理，但其中的hb_time字段显示为datetime.datetime(2022, 3, 15, 12, 56, 19)这样的格式，请问如何去除datetime.datetime标识，仅显示时间戳内容？对应的Row数据如下：

[Row(unit_id='00001597', country='ITA', gateway_id='8988228', first_hb_info=Row(hb_time=datetime.datetime(2022, 3, 15, 12, 56, 19), battery=60, ctrl=4, service=24, rssi=11, power=1, op_mode='IDL'), last_hb_info=Row(hb_time=datetime.datetime(2023, 11, 2, 20, 18, 21), battery=100, ctrl=4, service=27, rssi=10, power=1, op_mode='\x00\x00\x00'), last_op_mode_hb_info=Row(hb_time=datetime.datetime(2023, 11, 2, 20, 10, 21), battery=100, ctrl=4, service=24, rssi=10, power=1, op_mode='IDL')]

解决方法 #

方法一：Spark DataFrame阶段预处理（推荐）

优先在分布式计算阶段处理，避免把大量数据拉到Driver端影响性能。通过Spark内置函数将datetime字段转换为字符串或Unix时间戳格式：

转成可读字符串格式（如YYYY-MM-DD HH:mm:ss）

from pyspark.sql.functions import date_format, col, struct

# 依次处理三个嵌套的hb_time字段
df = df.withColumn(
    "first_hb_info",
    struct(
        date_format(col("first_hb_info.hb_time"), "yyyy-MM-dd HH:mm:ss").alias("hb_time"),
        col("first_hb_info.battery"),
        col("first_hb_info.ctrl"),
        col("first_hb_info.service"),
        col("first_hb_info.rssi"),
        col("first_hb_info.power"),
        col("first_hb_info.op_mode")
    )
).withColumn(
    "last_hb_info",
    struct(
        date_format(col("last_hb_info.hb_time"), "yyyy-MM-dd HH:mm:ss").alias("hb_time"),
        col("last_hb_info.battery"),
        col("last_hb_info.ctrl"),
        col("last_hb_info.service"),
        col("last_hb_info.rssi"),
        col("last_hb_info.power"),
        col("last_hb_info.op_mode")
    )
).withColumn(
    "last_op_mode_hb_info",
    struct(
        date_format(col("last_op_mode_hb_info.hb_time"), "yyyy-MM-dd HH:mm:ss").alias("hb_time"),
        col("last_op_mode_hb_info.battery"),
        col("last_op_mode_hb_info.ctrl"),
        col("last_op_mode_hb_info.service"),
        col("last_op_mode_hb_info.rssi"),
        col("last_op_mode_hb_info.power"),
        col("last_op_mode_hb_info.op_mode")
    )
)

# 再执行collect，得到的Row中hb_time就是字符串格式
processed_rows = df.collect()

转成Unix时间戳（秒级数值）

只需将上述代码中的date_format替换为unix_timestamp即可：

from pyspark.sql.functions import unix_timestamp, col, struct

# 以first_hb_info为例，其余字段处理逻辑一致
df = df.withColumn(
    "first_hb_info",
    struct(
        unix_timestamp(col("first_hb_info.hb_time")).alias("hb_time"),
        col("first_hb_info.battery"),
        col("first_hb_info.ctrl"),
        col("first_hb_info.service"),
        col("first_hb_info.rssi"),
        col("first_hb_info.power"),
        col("first_hb_info.op_mode")
    )
)

方法二：collect后在Python本地处理Row对象

如果已经将数据collect到本地，可以遍历Row对象，将嵌套的datetime字段转换为目标格式（Row不可变，需转成字典修改后重新构建Row）：

from pyspark.sql import Row

def format_row_datetime(row):
    # 处理第一个嵌套Row
    first_hb_dict = row.first_hb_info.asDict()
    # 转成字符串格式
    first_hb_dict["hb_time"] = first_hb_dict["hb_time"].strftime("%Y-%m-%d %H:%M:%S")
    # 或者转成Unix时间戳：first_hb_dict["hb_time"] = int(first_hb_dict["hb_time"].timestamp())
    
    # 处理第二个嵌套Row
    last_hb_dict = row.last_hb_info.asDict()
    last_hb_dict["hb_time"] = last_hb_dict["hb_time"].strftime("%Y-%m-%d %H:%M:%S")
    
    # 处理第三个嵌套Row
    last_op_hb_dict = row.last_op_mode_hb_info.asDict()
    last_op_hb_dict["hb_time"] = last_op_hb_dict["hb_time"].strftime("%Y-%m-%d %H:%M:%S")
    
    # 构建新的Row返回
    return Row(
        unit_id=row.unit_id,
        country=row.country,
        gateway_id=row.gateway_id,
        first_hb_info=Row(**first_hb_dict),
        last_hb_info=Row(**last_hb_dict),
        last_op_mode_hb_info=Row(**last_op_hb_dict)
    )

# 假设original_rows是collect得到的原始Row列表
processed_rows = [format_row_datetime(r) for r in original_rows]

处理后，hb_time字段会显示为2022-03-15 12:56:19这样的字符串，或对应的Unix时间戳数值，不再带有datetime.datetime标识。

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