Apache Beam中如何将空DeferredDataFrame转换为PCollection？

问题根源 #

当左反连接返回空的DeferredDataFrame时，pandas会丢失原始列的类型信息——空DataFrame的列会被自动推断为包含NaN的float类型（因为NaN属于float），而原左表的列类型是integer。Beam在将这样的空DataFrame转为PCollection时，会尝试将float类型的NaN转换为integer，从而触发ValueError: cannot convert float NaN to integer错误。

解决方案 #

修改左反连接函数，确保空结果返回的DataFrame保留原左表的列结构和数据类型，而不是直接返回过滤后的空DF。具体做法是：当过滤结果为空时，返回原左表的0行切片（left.iloc[0:0]），这个切片会完整保留原表的列名、数据类型等元信息。

修改后的代码 #

def left_anti_join(
    left: DeferredDataFrame,
    right: DeferredDataFrame,
    left_index: str,
    right_index: str, ) -> DeferredDataFrame:

    inner_joined_df = left.merge(
        right.set_index(right_index),
        right_index=True,
        left_on=left_index,
        how="left",
        indicator=True,
    )
    filtered_df = inner_joined_df[inner_joined_df._merge == "left_only"].drop("_merge", axis=1)
    
    # 空结果时返回原左表的0行切片，保留原始列类型
    return filtered_df if not filtered_df.empty else left.iloc[0:0]

额外优化（可选） #

如果需要更明确地控制列类型，也可以在返回空DF时显式指定类型，但这种方式需要手动维护列类型，不如直接复用原表结构高效：

# 仅当原左表列类型明确时使用
empty_df = pd.DataFrame(columns=left.columns, dtype=left.dtypes.to_dict())
return filtered_df if not filtered_df.empty else empty_df

为什么这个方案高效？ #

filtered_df.empty是DeferredDataFrame的延迟计算属性，不会触发额外的数据扫描或统计，比提前调用count()检查元素数量的方式性能更优，同时从根源上避免了类型转换错误。

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