如何在Pandas中对比两张表并基于同主键日期比较筛选行

• 语法混用：pandas DataFrame 不支持 PySpark 的desc()语法，两者API完全不兼容
• 操作未生效：pandas的drop_duplicates、sort_values（你原来写的sort是错误的pandas方法）默认返回新对象，没有赋值给变量等于操作没有执行
• 比较逻辑错误：直接对两个排序后的全表做大小对比，既没有按主键匹配对应记录，也没有逐行对比日期，完全不符合需求
• 日期格式风险：如果日期字段是字符串格式，直接比较会出现逻辑错误，需要先转换为datetime类型

我们的目标逻辑是：两张表按主键visit_occurrence_id关联，筛选表2（住院表）中visit_start_date大于表1（门诊表）对应记录日期的行，输出新表。

Pandas 实现（匹配你原有转pd.DataFrame的逻辑） #

import pandas as pd

def fluvoxamine_covid_hospital_progression(fluvoxamine_covids_that_were_also_in_hospital_at_any_time, fluvoxamine_for_hospitalization_analysis_only_outpatients_need_to_dbl_chk):
    # 处理表1：门诊表
    df_outpatient = pd.DataFrame(fluvoxamine_for_hospitalization_analysis_only_outpatients_need_to_dbl_chk)
    # 按主键去重、日期转格式、只保留需要的主键和日期字段，避免合并字段冗余
    df_outpatient = df_outpatient.drop_duplicates(['visit_occurrence_id'])[['visit_occurrence_id', 'visit_start_date']]
    df_outpatient['visit_start_date'] = pd.to_datetime(df_outpatient['visit_start_date'])
    # 重命名日期字段避免合并后重名冲突
    df_outpatient = df_outpatient.rename(columns={'visit_start_date': 'outpatient_date'})

    # 处理表2：住院表
    df_hospital = pd.DataFrame(fluvoxamine_covids_that_were_also_in_hospital_at_any_time)
    df_hospital = df_hospital.drop_duplicates(['visit_occurrence_id'])
    df_hospital['visit_start_date'] = pd.to_datetime(df_hospital['visit_start_date'])

    # 按主键关联两张表
    df_merge = pd.merge(df_hospital, df_outpatient, on='visit_occurrence_id', how='inner')

    # 筛选住院日期晚于门诊日期的记录，保留住院表的原有字段
    df_result = df_merge[df_merge['visit_start_date'] > df_merge['outpatient_date']][df_hospital.columns]

    # 可根据需要写入文件，示例：df_result.to_csv('result.csv', index=False)
    return df_result

PySpark 实现（适合大数据量场景，匹配你原有导入的pyspark包） #

如果你的原始数据是Spark DataFrame，不需要转pandas，直接用Spark API实现性能更高：

from pyspark.sql import functions as F

def fluvoxamine_covid_hospital_progression(spark_df_hospital, spark_df_outpatient):
    # 按主键去重、日期转格式
    df_outpatient = spark_df_outpatient.dropDuplicates(['visit_occurrence_id'])\
        .withColumn('outpatient_date', F.to_date(F.col('visit_start_date')))\
        .select('visit_occurrence_id', 'outpatient_date')
    
    df_hospital = spark_df_hospital.dropDuplicates(['visit_occurrence_id'])\
        .withColumn('visit_start_date', F.to_date(F.col('visit_start_date')))
    
    # 关联+筛选
    df_result = df_hospital.join(df_outpatient, on='visit_occurrence_id', how='inner')\
        .filter(F.col('visit_start_date') > F.col('outpatient_date'))\
        .select(df_hospital.columns)
    
    # 可根据需要写入新表，示例：df_result.write.saveAsTable('new_result_table')
    return df_result

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