pyspark：ValueError：推断后无法确定某些类型答案

【问题标题】：pyspark: ValueError: Some of types cannot be determined after inferringpyspark：ValueError：推断后无法确定某些类型
【发布时间】：2023-04-05 05:38:01
【问题描述】：

我有一个熊猫数据框my_df，my_df.dtypes 给了我们：

ts              int64
fieldA         object
fieldB         object
fieldC         object
fieldD         object
fieldE         object
dtype: object

然后我尝试通过以下操作将 pandas 数据框 my_df 转换为 spark 数据框：

spark_my_df = sc.createDataFrame(my_df)

但是，我收到以下错误：

ValueErrorTraceback (most recent call last)
<ipython-input-29-d4c9bb41bb1e> in <module>()
----> 1 spark_my_df = sc.createDataFrame(my_df)
      2 spark_my_df.take(20)

/usr/local/spark-latest/python/pyspark/sql/session.py in createDataFrame(self, data, schema, samplingRatio)
    520             rdd, schema = self._createFromRDD(data.map(prepare), schema, samplingRatio)
    521         else:
--> 522             rdd, schema = self._createFromLocal(map(prepare, data), schema)
    523         jrdd = self._jvm.SerDeUtil.toJavaArray(rdd._to_java_object_rdd())
    524         jdf = self._jsparkSession.applySchemaToPythonRDD(jrdd.rdd(), schema.json())

/usr/local/spark-latest/python/pyspark/sql/session.py in _createFromLocal(self, data, schema)
    384 
    385         if schema is None or isinstance(schema, (list, tuple)):
--> 386             struct = self._inferSchemaFromList(data)
    387             if isinstance(schema, (list, tuple)):
    388                 for i, name in enumerate(schema):

/usr/local/spark-latest/python/pyspark/sql/session.py in _inferSchemaFromList(self, data)
    318         schema = reduce(_merge_type, map(_infer_schema, data))
    319         if _has_nulltype(schema):
--> 320             raise ValueError("Some of types cannot be determined after inferring")
    321         return schema
    322 

ValueError: Some of types cannot be determined after inferring

有谁知道上面的错误是什么意思？谢谢！

【问题讨论】：

标签： python python-2.7 pandas pyspark spark-dataframe

【解决方案1】：

如果你使用RDD[Row].toDF()monkey-patched方法，你可以增加采样率，在推断类型时检查超过100条记录：

# Set sampleRatio smaller as the data size increases
my_df = my_rdd.toDF(sampleRatio=0.01)
my_df.show()

假设您的 RDD 中的所有字段中都有非空行，当您将 sampleRatio 增加到 1.0 时，它更有可能找到它们。

【讨论】：

如果你的 rdd 非常大，让你的采样率更像 0.01，否则 spark 在工作的最后会花费很长时间
@crypdick 我会修改答案，这是一个更好的默认值，谢谢。

【解决方案2】：

我也遇到过同样的问题，如果您不需要为 null 的列，您可以在导入到 spark 之前简单地将它们从 pandas 数据框中删除：

my_df = my_df.dropna(axis='columns', how='all') # Drops columns with all NA values
spark_my_df = sc.createDataFrame(my_df)

【讨论】：

如果不是从 pandas 导入，你会怎么做？
这取决于您使用什么来导入，最初的问题是关于从 Pandas 导入。

【解决方案3】：

这可能是因为所有列都具有空值。您应该先删除这些列，然后再将它们转换为 spark 数据框

【讨论】：

【解决方案4】：

为了推断字段类型，PySpark 会查看每个字段中的非无记录。如果一个字段只有 None 记录，PySpark 无法推断类型并会引发该错误。

手动定义架构将解决问题

>>> from pyspark.sql.types import StructType, StructField, StringType
>>> schema = StructType([StructField("foo", StringType(), True)])
>>> df = spark.createDataFrame([[None]], schema=schema)
>>> df.show()
+----+
|foo |
+----+
|null|
+----+

【讨论】：

我可以只给出整个 None 列的架构并跳过其余列吗？

【解决方案5】：

要解决此问题，您可以提供自己定义的架构。

例如：

重现错误：

>>> df = spark.createDataFrame([[None, None]], ["name", "score"])

修复错误：

>>> from pyspark.sql.types import StructType, StructField, StringType, DoubleType
>>> schema = StructType([StructField("name", StringType(), True), StructField("score", DoubleType(), True)])
>>> df = spark.createDataFrame([[None, None]], schema=schema)
>>> df.show()
+----+-----+
|name|score|
+----+-----+
|null| null|
+----+-----+

【讨论】：

如果我们有超过 2 列，并且只有 1 列完全为空，是否有更好的优雅方式来传递架构，而无需为所有列显式定义架构？