如何在使用pyspark的条件下使用for循环？答案

【问题标题】：How to use for loop in when condition using pyspark?如何在使用pyspark的条件下使用for循环？
【发布时间】：2019-11-20 07:33:29
【问题描述】：

我正在尝试检查多个列值是否为0。我们有从 1 到 11 列的 spark 数据框，需要检查它们的值。目前我的代码如下所示：-

df3 =df3.withColumn('Status', when((col("1") ==0)|(col("2") ==0)|(col("3") ==0)| (col("4") ==0) |(col("5") ==0)|(col("6") ==0)|(col("7") ==0)| (col("8") ==0)|(col("9") ==0)|(col("10") ==0)| (col("11") ==0) ,'Incomplete').otherwise('Complete'))

我怎样才能通过只使用 for 循环而不是这么多 or 条件来实现这一点

【问题讨论】：

你的列真的只用数字命名吗？这是一个可怕的名字。通过正确的命名（至少 c1、c2 等...），您可以简单地使用 F.expr 来获得结果。

标签： pyspark

【解决方案1】：

我提出了一个更pythonic 的解决方案。使用functools.reduce 和operator.or_。

import operator
import functools

colnames = [str(i+1) for i in range(11)]
df1 = spark._sc.parallelize([
  [it for it in range(11)], 
  [it for it in range(1,12)]]
).toDF((colnames))

df1.show()
+---+---+---+---+---+---+---+---+---+---+---+
|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11|
+---+---+---+---+---+---+---+---+---+---+---+
|  0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10|
|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11|
+---+---+---+---+---+---+---+---+---+---+---+

cond_expr = functools.reduce(operator.or_, [(f.col(c) == 0) for c in df1.columns])

df1.withColumn('test', f.when(cond_expr, f.lit('Incomplete')).otherwise('Complete')).show()
+---+---+---+---+---+---+---+---+---+---+---+----------+
|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11|      test|
+---+---+---+---+---+---+---+---+---+---+---+----------+
|  0|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10|Incomplete|
|  1|  2|  3|  4|  5|  6|  7|  8|  9| 10| 11|  Complete|
+---+---+---+---+---+---+---+---+---+---+---+----------+

这样您就不需要定义任何函数、评估字符串表达式或使用 python lambdas。希望这可以帮助。

【讨论】：

【解决方案2】：

您可以使用下面的代码收集您的条件并将它们连接成一个字符串，然后调用eval。

代码

cond ='|'.join('(col("'+str(_)+'")==0)' for _ in range(1, 12))

cond = '('+cond+')'

print(cond)

#((col("1")==0)|(col("2")==0)|(col("3")==0)|(col("4")==0)|(col("5")==0)|(col("6")==0)|(col("7")==0)|(col("8")==0)|(col("9")==0)|(col("10")==0)|(col("11")==0))

df3 = df3.withColumn('Status', when(eval(cond),'Incomplete').otherwise('Complete'))

【讨论】：

@Amol 欢迎您。如果您愿意，您也可以研究其他更好的解决方案。

【解决方案3】：

可能有更好的解决方案

>>> df = spark.createDataFrame([(1,0,0,2),(1,1,1,1)],['c1','c2','c3','c4'])
>>> df.show()
+---+---+---+---+
| c1| c2| c3| c4|
+---+---+---+---+
|  1|  0|  0|  2|
|  1|  1|  1|  1|
+---+---+---+---+

def status(x):
  l = [i for i in x]
  if 0 in l:
    return 'Incomplete'
  else:
    return 'Complete'

>>> df.rdd.map(lambda x:  (x.c1, x.c2, x.c3, x.c4,status(x))).toDF(['c1','c2','c3','c4','status']).show()
+---+---+---+---+----------+
| c1| c2| c3| c4|    status|
+---+---+---+---+----------+
|  1|  0|  0|  2|Incomplete|
|  1|  1|  1|  1|  Complete|
+---+---+---+---+----------+

【讨论】：

虽然这将在一个小示例中起作用，但这并不能真正扩展，因为 rdd.map 和 lambda 的组合将强制 Spark 驱动程序为 status() 回调 python功能并失去并行化的好处。