Pyspark：如何以数据框的行作为参数应用用户定义的函数？

Question

我有一个包含 87 列的 PySpark 数据框。我想将数据框的每一行传递给一个函数并为每一行获取一个列表，以便我可以单独创建一个列。

PySpark 代码

UDF：

def make_range_vector(row,categories,ledger):
    print(type(row),type(categories),type(ledger))                
    category_vector=[]
    for category in categories:
      if(row[category]!=0):
         category_percentage=func.round(row[category]*100/row[ledger])
         category_vector.append(category_percentage)
      else:
          category_vector.append(0)
    category_vector=sqlCtx.createDataFrame(category_vector,IntegerType())    
    return category_vector

主要功能

pivot_card.withColumn('category_debit_vector',(make_range_vector(struct([pivot_card[x]  for x in pivot_card.columns] ),pivot_card.columns[3:],'debit')))

我是 PySpark 的初学者，我无法找到以下问题的答案。

1. if(row[category]!=0): 这个语句给了我 ValueError: ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.

2. 所以，我在函数内部打印了参数。 它输出<class 'pyspark.sql.column.Column'> <class 'list'> <class #'str'>。不应该是StructType吗？

3. 我可以像在 Pandas 中那样传递一个 Row 对象并执行类似的操作吗？

我查看了很多来源，大部分来自这个问题和这个来源 (https://community.hortonworks.com/questions/130866/rowwise-manipulation-of-a-dataframe-in-pyspark.html)

PySpark row-wise function composition

Answer 1

我发现了我在代码中犯的愚蠢错误。我没有调用 UDF，而是调用了原始函数。已在以下答案中更正：

主要功能

pivot_card.withColumn('category_debit_vector',(make_range_vector_udf(struct([pivot_card[x] for x in pivot_card.columns] ),pivot_card.columns[3:],'debit')))

编辑

我知道我们不能真正在 UDF 中传递其他参数。 谢谢。