PySpark 安全列类型转换答案

【问题标题】：PySpark safe column type conversionPySpark 安全列类型转换
【发布时间】：2017-09-29 21:02:06
【问题描述】：

如果我将列转换为不同的类型，那么任何无法转换的数据都会被静默转换为 NULL：

df = sc.parallelize([('a', '1'), ('b', 'NAN'), ('c', 'foo'))]) .toDF(['name', 'values'])

df.select( df.values.cast(DoubleType()) ).show()
+------+
|values|
+------+
|   1.0|
|  null|
|  null|
+------+

如果任何值无法转换，有什么方法可以让转换产生错误？

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

您可以使用 python 的 float 函数为此编写自己的 udf，该函数在字符串无法转换为浮点数时引发错误：

import pyspark.sql.functions as F
@F.udf
def cast_safe(x):
    return float(x)

df.select(cast_safe(df.values).alias("values")).show()

[Stage 5:> (0 + 3) / 3]17/09/29 09:00:29 ERROR Executor: Exception in task 2.0 in 阶段 5.0 (TID 11) org.apache.spark.api.python.PythonException：回溯（最近一次调用最后一次）：文件“C:\spark\spark-2.0.1\python\lib\pyspark.zip\pyspark\worker.py”，行 172，主要文件“C:\spark\spark-2.0.1\python\lib\pyspark.zip\pyspark\worker.py”，行 167，处理中文件“C:\spark\spark-2.0.1\python\lib\pyspark.zip\pyspark\worker.py”，行 106，在文件“C:\spark\spark-2.0.1\python\lib\pyspark.zip\pyspark\worker.py”，行 92，在文件“C:\spark\spark-2.0.1\python\lib\pyspark.zip\pyspark\worker.py”，行 70，在文件“”，第 3 行，在 cast_safe 中 ValueError：无法将字符串转换为浮点数：foo

如果列中的所有元素都可以安全地转换，则可以正常工作：

df1 = sc.parallelize([('a', '1'), ('b', '0'), ('c', '2.')]) .toDF(['name', 'values'])
df1.select(df1.name, cast_safe(df1.values).alias("values")).show()
+----+------+
|name|values|
+----+------+
|   a|   1.0|
|   b|   0.0|
|   c|   2.0|
+----+------+

【讨论】：

虽然这确实有效，但我宁愿避免为每个浮点数调用 python 函数。