根据其他列中满足的条件添加列

Question

我是PySpark的新手，目前在以下问题上将面临挑战。我有一个火花df，如下所示

DeviceID     max(A)    max(B)    max(INUT)
0023002      2.5       3.7       8.1
0023045      2.2       1.3       11.3
0023008      4.7       2.3       1.9

如何将另一列添加为“状态”，其中的值将基于以下逻辑。

if 0.20 * max(INUT) > max(max(A),max(B)) then Status = 'Imbalance' else 'Balance'

预期上述逻辑将产生以下数据。

DeviceID     max(A)    max(B)    max(INUT)    Status
0023002      2.5       3.7       8.1          'Balance'
0023045      2.2       1.3      11.3          'ImBalance'
0023008      4.7       2.3       1.9          'Balance'

现在要实现上面的df，下面是我正在使用的代码

from pyspark.sql.function import col
import pyspark.sql.function as F
df_final = df.withColumn(
             'Status',
             F.when(col('max(INUT)')*0.20 > F.greatest(col('max(A)'),col('max(B)'),
             'Imbalance')\
         .otherwise('Balance')

上面的代码段正在抛出一个错误

AttributeError: 'tuple' object has no attribute 'otherwise'

我错过了哪里？如有任何提示，将不胜感激。

Answer 1

这里有一些小语法错误，这是您的最后代码：

import pyspark.sql.functions as F

df = spark.createDataFrame(
[("0023002", 2.5, 3.7, 8.1),
("0023045", 2.2, 1.3, 11.3),
("0023008", 4.7, 2.3, 1.9)], ["DeviceID", "max_A", "max_B", "max_INUT"])

df_final = df.withColumn('Status', \
             F.when(F.col('max_INUT')*0.20 > F.greatest(F.col('max_A'),F.col('max_B')), 'Imbalance') \
         .otherwise('Balance'))

和一些评论/评论：

1. 要使用pyspark.sql.functions中的函数，只需使用F别名即可。你不需要进口它两次。
2. 有一些遗漏的括号
3. 我还替换了max(A) -> max_A，因为它使我更容易阅读，我相信

输出：

+--------+-----+-----+--------+---------+
|DeviceID|max_A|max_B|max_INUT|   Status|
+--------+-----+-----+--------+---------+
| 0023002|  2.5|  3.7|     8.1|  Balance|
| 0023045|  2.2|  1.3|    11.3|Imbalance|
| 0023008|  4.7|  2.3|     1.9|  Balance|
+--------+-----+-----+--------+---------+