замените число с плавающей точкой на нули в PySpark

Question

У меня были некоторые значения, которые отображались в моем фрейме данных PySpark как NaN, и я обнаружил, что могу преобразовать их в значения NULL. Затем я корректировал эти значения NULL, вменяя это значение в нечто другое. При этом я обнаружил, что он также превращает 0 в ряде моих столбцов в NULL. Почему это происходит, и как я могу преобразовать nans в NULL, не влияя на 0s?

cSchema = StructType([StructField("col", LongType())])
vals = [[0] for i in range(20)]
test_df = spark.createDataFrame(vals,schema=cSchema)

test_df.show(20)

+---+
|col|
+---+
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|
|  0|

test_df = test_df.replace(float('nan'), None)

test_df.show(20)

+----+
| col|
+----+
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|
|null|

Answer 1

Схема в вашем примере не подходит для операции, которую вы пытаетесь выполнить. Вы ищете значение с плавающей точкой в ​​столбце (длинных) целых чисел. Я удивлен, replace не игнорирует колонку в целом ...
Вот что происходит, когда вы пытаетесь создать такой DF напрямую:

>>> cSchema = StructType([StructField("col1", LongType()),StructField("col2", LongType())])
... vals = [[0, float('nan')] for i in range(20)]
... test_df = spark.createDataFrame(vals,schema=cSchema)
...
... test_df.show(20)
Traceback (most recent call last):
  File "<stdin>", line 3, in <module>
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\session.py", line 748, in createDataFrame
    rdd, schema = self._createFromLocal(map(prepare, data), schema)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\session.py", line 413, in _createFromLocal
    data = list(data)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\session.py", line 730, in prepare
    verify_func(obj)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\types.py", line 1389, in verify
    verify_value(obj)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\types.py", line 1370, in verify_struct
    verifier(v)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\types.py", line 1389, in verify
    verify_value(obj)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\types.py", line 1383, in verify_default
    verify_acceptable_types(obj)
  File "D:\Spark\spark-2.4.4-bin-hadoop2.7\python\pyspark\sql\types.py", line 1278, in verify_acceptable_types
    % (dataType, obj, type(obj))))
TypeError: field col2: LongType can not accept object nan in type <class 'float'>

field col2: LongType can not accept object nan in type <class 'float'>

А вот что происходит, когда вы используете соответствующую схему

>>> cSchema = StructType([StructField("col1", DoubleType()),StructField("col2", DoubleType())])
... vals = [[0., float('nan')] for i in range(20)]
... test_df = spark.createDataFrame(vals,schema=cSchema)
...
... test_df.show(3)
+----+----+
|col1|col2|
+----+----+
| 0.0| NaN|
| 0.0| NaN|
| 0.0| NaN|
+----+----+
only showing top 3 rows

>>> test_df.replace(float('nan'), None).show(3)
+----+----+
|col1|col2|
+----+----+
| 0.0|null|
| 0.0|null|
| 0.0|null|
+----+----+
only showing top 3 rows

Так что либо попробуйте привести все заранее к плавающему / двойному числу (если nan-s смешаны в целочисленном столбце), либо используйте параметр subset replace, чтобы укажите только плавающие столбцы для поиска.