将带有日期的Spark数据框转换为Pandas数据框时出现错误

Question

我有一个带有以下架构的 Spark DataFrame：

root
 |-- product_id: integer (nullable = true)
 |-- stock: integer (nullable = true)
 |-- start_date: date (nullable = true)
 |-- end_date: date (nullable = true)

当尝试将其传递给 pandas_udf 或使用以下方式转换为pandas dataframe时：

pandas_df = spark_df.toPandas()

它返回以下错误信息：

AttributeError        Traceback (most recent call last)
<ipython-input-86-4bccc6e8422d> in <module>()
     10 # spark_df.printSchema()
     11 
---> 12 pandas_df = spark_df.toPandas()

/home/.../lib/python2.7/site-packages/pyspark/sql/dataframe.pyc in toPandas(self)
   2123                         table = pyarrow.Table.from_batches(batches)
   2124                         pdf = table.to_pandas()
-> 2125                         pdf = _check_dataframe_convert_date(pdf, self.schema)
   2126                         return _check_dataframe_localize_timestamps(pdf, timezone)
   2127                     else:

/home.../lib/python2.7/site-packages/pyspark/sql/types.pyc in _check_dataframe_convert_date(pdf, schema)
   1705     """
   1706     for field in schema:
-> 1707         pdf[field.name] = _check_series_convert_date(pdf[field.name], field.dataType)
   1708     return pdf
   1709 

/home/.../lib/python2.7/site-packages/pyspark/sql/types.pyc in _check_series_convert_date(series, data_type)
   1690     """
   1691     if type(data_type) == DateType:
-> 1692         return series.dt.date
   1693     else:
   1694         return series

/home/.../lib/python2.7/site-packages/pandas/core/generic.pyc in __getattr__(self, name)
   5061         if (name in self._internal_names_set or name in self._metadata or
   5062                 name in self._accessors):
-> 5063             return object.__getattribute__(self, name)
   5064         else:
   5065             if self._info_axis._can_hold_identifiers_and_holds_name(name):

/home/.../lib/python2.7/site-packages/pandas/core/accessor.pyc in __get__(self, obj, cls)
    169             # we're accessing the attribute of the class, i.e., Dataset.geo
    170             return self._accessor
--> 171         accessor_obj = self._accessor(obj)
    172         # Replace the property with the accessor object. Inspired by:
    173         # http://www.pydanny.com/cached-property.html

/home/.../lib/python2.7/site-packages/pandas/core/indexes/accessors.pyc in __new__(cls, data)
    322             pass  # we raise an attribute error anyway
    323 
--> 324         raise AttributeError("Can only use .dt accessor with datetimelike "
    325                              "values")

AttributeError: Can only use .dt accessor with datetimelike values

如果从Spark DataFrame中删除日期字段，则转换可以顺利进行。我已经检查了数据不包含任何空值，但是知道如何处理这些内容会更好。我使用的是Python 2.7，并且安装了以下软件包：pyspark==2.4.0、pyarrow==0.12.1、pandas==0.24.1。

- Unaipg

---

如果可以的话，请将您的日期字段转换为DateType或Timestamp类型。 - Victor Valente

@VictorValente 你说的 DateType 是什么意思？难道这个字段不是已经在模式中以此类型显示了吗？ - victorx

Answer 1

看起来像是一个bug。我在使用pyarrow==0.12.1和pyarrow==0.12.0时遇到了相同的问题。将spark dataframe列转换为时间戳对我有用。

spark.sql('SELECT CAST(date_column as TIMESTAMP) FROM foo')

回滚到pyarrow==0.11.0也可以解决这个问题。(我的Python版本是3.7.1和pandas版本是0.24.2)

- Aska Fed

---

仍然发生在pyarrow==0.14.0和spark 2.4.0中。 - Mithril

使用pyspark 2.4.3和pyarrow 0.17.0时出现相同的问题。 - victorx

Answer 2

根据 Jira，他们在 Spark 3 中修复了该问题。作为解决方法，您可以考虑将日期列转换为时间戳（这与 pandas datetime 类型更加对齐）。

import pyspark.sql.functions as func
df = df.select(func.to_timestamp(func.col('session_date'), 'yyyy-MM-dd').alias('session_date')
df.toPandas()

在Pyspark 2.4.4中测试通过

- Ricardo Mutti

Answer 3

这对我有用：

这对我有用：

import pyspark.sql.functions as f

spark_df = spark_df.withColumn('start_date', f.to_timestamp(f.col('start_date')))
spark_df = spark_df.withColumn('end_date',   f.to_timestamp(f.col('end_date')))
pandas_df = spark_df.toPandas()

- ChrisDanger