如何使用WHEN子句检查Pyspark数据帧的String列上的Null条件?

我正在试着检查 NULL 或 empty string 在数据帧的字符串列上,以及 0 对于下面给出的整数列。

emp_ext = emp_ext.withColumn('emp_header', when((F.col('emp_header').isNull()) | (F.col('emp_header') == '0'), 'UNKNOWN')) \
    .withColumn('emp_item', when((F.col('emp_item').isNull()) | (F.col('emp_item') == 0), -1)) \
    .withColumn('emp_lease', when((F.col('emp_header').isNull() | F.col('emp_header') == '0') & (F.col('emp_item').isNull() | F.col('emp_item') == 0), -1)))

该列 emp_header 是字符串列, emp_item 是Integer列,并且 emp_lease 是Integer列。

当我运行上面的代码时,我得到一个错误,说列中存在数据类型不匹配 emp_header 之间 NULL & STRING 如下所示。

2022-03-03 07:17:41,931 - src.emp_load -   76 - ERROR - Failed to load history data into emp_data table with the exception: cannot resolve '((`emp_header` IS NULL) OR `emp_header`)' due to data type mismatch: differing types in '((`emp_header` IS NULL) OR `emp_header`)' (boolean and string).;;

因此,我尝试将NULL转换为不同的语法,如下所示。

emp_ext = emp_ext.withColumn('emp_header', when((F.col('emp_header') == '') | (F.col('emp_header') == '0'), 'UNKOWN')) \
    .withColumn('emp_item', when((F.col('emp_item') == '') | (F.col('emp_item') == 0), -1)) \
    .withColumn('emp_lease', when((F.col('emp_header') == '' | F.col('emp_header') == '0') & (F.col('emp_item') == '' | F.col('emp_item') == 0), -1))

这一次,我的Pycharm上的编译器说: Expected type 'Column', got 'str' instead 在下面屏幕截图中标记的行。

如果我继续运行带有上述更改的代码,代码将失败,并出现不同的异常

2022-03-03 07:46:08,227 - src.emp_load - 76 - ERROR - Failed to load history data into emp_table with the exception: An error occurred while calling o336.or. Trace:
py4j.Py4JException: Method or([class java.lang.String]) does not exist
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:251)
    at java.lang.Thread.run(Thread.java:748)

Traceback (most recent call last):
  File "/local_disk0/tmp/spark-a2890f41-1167-481a-85a9-6984c04d05c2/template_python-1.0.0-py3-none-any.whl/src/emp_load.py", line 71, in main
    so_lookup_insert(spark=spark, df=df, years=years, columns=columns)
  File "/local_disk0/tmp/spark-a2890f41-1167-481a-85a9-6984c04d05c2/template_python-1.0.0-py3-none-any.whl/src/emp_load.py", line 100, in so_lookup_insert
    .withColumn('emp_lease', when((F.col('emp_header') == '' | F.col('emp_header') == '0') & (F.col('emp_item') == '' | F.col('emp_item') == 0), -1))
  File "/databricks/spark/python/pyspark/sql/column.py", line 118, in _
    njc = getattr(self._jc, name)(jc)
  File "/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1305, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/databricks/spark/python/pyspark/sql/utils.py", line 127, in deco
    return f(*a, **kw)
  File "/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py", line 332, in get_return_value
    format(target_id, ".", name, value))

更新1: 我在 when 正如评论中所建议的,第三行的条件,我不再面临第二个例外。 我做了 isNull() 到 '' 如下所示。

emp_ext = emp_ext.withColumn('emp_header', when((F.col('emp_header') == '') | (F.col('emp_header') == '0'), 'UNKNOWN')) \
    .withColumn('emp_item', when((F.col('emp_item') == '') | (F.col('emp_item') == 0), -1)) \
    .withColumn('emp_lease', when(((F.col('emp_header') == '') | F.col('emp_header') == '0') & ((F.col('emp_item') == '') | F.col('emp_item') == 0), -1)))

但我仍然看到了例外:

2022-03-03 08:41:31,295 - src.emp_load - 76 - ERROR - Failed to load history data into emp_table with the exception: cannot resolve '((`emp_header` = '') OR `emp_header`)' due to data type mismatch: differing types in '((`emp_header` = '') OR `emp_header`)' (boolean and string).;;

有人能告诉我我在这里犯的错误是什么吗?我该如何解决?