如何将Spyder中的Python条件语句转换为Databricks PySpark实现

需求背景 #

需要将Spyder中编写的Python if/else条件逻辑，转换为Azure Databricks上的纯PySpark代码。原逻辑为：判断DataFrame行数是否大于0，若大于0则筛选出idnum不在历史表中的数据；否则直接复用另一张DataFrame。尝试过PySpark的When/Otherwise但报错（因该方法针对列级操作），且要求使用纯Python实现，不借助%sql魔法命令。

原Python代码 #

length_df = df.shape[0]
if(length_df>0):
    qry='''
    select *
    from df where idnum not in
    (select idnum from df_Prev)
    '''
else:
    df_new= df_.copy()

解决思路与实现代码 #

1. 替换行数获取方式 #

原生Python用df.shape[0]取行数，PySpark中需用count()方法（注意：count()是行动算子，会触发Spark计算）：

length_df = df.count()

2. 用Python原生if/else做分支判断 #

这里的判断是整个数据集的行数条件，而非每行的列值判断，因此直接用Python的if/else即可，无需使用PySpark的When/Otherwise（后者仅适用于列级逐行判断）。

3. 用PySpark API替代SQL的not in逻辑 #

原SQL的not in逻辑，在PySpark中推荐用left_anti join实现（性能优于not in，尤其适合大数据场景），它会返回左表中在右表无匹配的行，完全对应原需求：

# 完整实现代码
length_df = df.count()

if length_df > 0:
    # left_anti join实现idnum不在df_Prev中的筛选
    df_new = df.join(df_Prev, on="idnum", how="left_anti")
else:
    # PySpark DataFrame是不可变对象，无需copy，直接赋值即可
    df_new = df_

关键说明 #

• PySpark DataFrame是不可变的，因此不需要copy()方法，直接赋值即可；若需缓存数据可使用df_.cache()。
• left_anti join是实现"不在某集合中"逻辑的最优方式，避免了not in在大数据量下的性能问题。
• 若仅需判断DataFrame是否为空，也可使用df.rdd.isEmpty()，同样会触发计算。

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