PySpark中使用UDF基于列值提取数据行的实现疑问

需要从DataFrame中提取columns_list = [column2, column3]中任意一列存在值的行，原始DataFrame如下：

column1       column2        column3
   abc           def             ghi
   abc1                           
   abc2              
   abc3          jkl
   abc4                          mno

期望得到的结果DataFrame：

column1       column2        column3
   abc           def             ghi       
   abc3          jkl
   abc4                          mno

尝试了以下自定义函数（UDF）但不清楚如何将行传入：

def valuesAny(string, array):
  if len(string) == 0:
    return False
  else:
    return True

推荐方案：用Pandas内置方法（无需自定义函数） #

直接利用Pandas的布尔判断和any()方法即可高效实现，无需自定义UDF：

1. 针对指定列，判断每行的每个值是否非空（假设空值为空白字符串，若为NaN可替换为pd.isna）
2. 使用any(axis=1)判断每行是否至少有一列满足非空条件
3. 用得到的布尔序列作为索引过滤原DataFrame

代码示例

import pandas as pd

# 构造原始DataFrame
df = pd.DataFrame({
    'column1': ['abc', 'abc1', 'abc2', 'abc3', 'abc4'],
    'column2': ['def', '', '', 'jkl', ''],
    'column3': ['ghi', '', '', '', 'mno']
})

columns_list = ['column2', 'column3']
# 过滤指定列中任意一列非空的行
result_df = df[(df[columns_list] != '').any(axis=1)]
print(result_df)

自定义函数实现方式 #

如果坚持要使用自定义函数，可通过apply方法按行传入数据：

1. 修改自定义函数，使其接收整行数据和指定列列表
2. 用df.apply(..., axis=1)按行应用函数，得到布尔过滤序列
3. 过滤原DataFrame得到结果

代码示例

import pandas as pd

# 构造原始DataFrame
df = pd.DataFrame({
    'column1': ['abc', 'abc1', 'abc2', 'abc3', 'abc4'],
    'column2': ['def', '', '', 'jkl', ''],
    'column3': ['ghi', '', '', '', 'mno']
})

columns_list = ['column2', 'column3']

def valuesAny(row, cols):
    # 检查指定列中是否存在非空值
    return any(row[col].strip() != '' for col in cols)

# axis=1表示按行应用函数，args传入额外参数
result_df = df[df.apply(valuesAny, args=(columns_list,), axis=1)]
print(result_df)

注意：若数据中的空值是NaN而非空白字符串，将判断条件替换为pd.notna(row[col])即可。

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