Pyspark: ValueError: некоторые типы не могут быть определены после вывода

У меня есть кадр данных pandas my_df, а my_df.dtypes дает нам:

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

Затем я пытаюсь преобразовать фрейм данных pandas my_df в рамку искровых данных, сделав ниже:

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

Кто-нибудь знает, что означает вышеупомянутая ошибка? Спасибо!

Ответ 1

Чтобы вывести тип поля, PySpark просматривает записи non-none в каждом поле. Если поле имеет только записи 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|
+----+

Ответ 2

И чтобы устранить эту проблему, вы можете предоставить свою собственную определенную схему.

Например:

Чтобы воспроизвести ошибку:

>>> 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|
+----+-----+

Ответ 3

Если вы используете обезьяно-исправленный метод RDD[Row].toDF() вы можете увеличить коэффициент выборки, чтобы проверить более 100 записей при выводе типов:

my_df = my_rdd.toDF(sampleRatio=0.1)
my_df.show()

Предполагая, что во всех полях вашего RDD есть ненулевые строки, будет более вероятно найти их, когда вы увеличите sampleRatio до 1.0.