PySpark转Pandas DataFrame报错：'DataFrame'对象无'dtype'属性

问题描述 #

在Databricks环境中执行PySpark DataFrame转Pandas DataFrame操作时，运行pd_flights = flights.select("*").toPandas()触发以下错误：

/databricks/spark/python/pyspark/sql/pandas/conversion.py:145: UserWarning: toPandas attempted Arrow optimization because 'spark.sql.execution.arrow.pyspark.enabled' is set to true, but has reached the error below and can not continue. Note that 'spark.sql.execution.arrow.pyspark.fallback.enabled' does not have an effect on failures in the middle of computation.
  'DataFrame' object has no attribute 'dtype'
  warnings.warn(msg)
AttributeError: 'DataFrame' object has no attribute 'dtype'

尝试通过spark.conf.set("spark.sql.execution.arrow.enabled", "false")关闭Arrow优化后，问题仍未解决。目标PySpark DataFrame的schema如下：

flight_id: string (nullable = true)
 |-- flight_direction: string (nullable = true)
 |-- service_type: string (nullable = true)
 |-- flight_designator: string (nullable = true)
 |-- flight_number: string (nullable = true)
 |-- callsign: string (nullable = true)
 |-- scheduled_datetime: timestamp (nullable = true)
 |-- connecting_flight_designator: string (nullable = true)
 |-- airport_iata_codes: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- airline_name: string (nullable = true)
 |-- airport_names: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- country_number: long (nullable = true)
 |-- eu_category: string (nullable = true)
 |-- safe_town_indicator: boolean (nullable = true)
 |-- sibt: timestamp (nullable = true)
 |-- aibt: timestamp (nullable = true)
 |-- sobt: timestamp (nullable = true)
 |-- aibt: timestamp (nullable = true)
 |-- tsat: timestamp (nullable = true)
 |-- aircraft_name: string (nullable = true)
 |-- aircraft_registration: string (nullable = true)
 |-- ramp: string (nullable = true)
 |-- ramp_previous: string (nullable = true)
 |-- seats: long (nullable = true)
 |-- actual_total_pax: integer (nullable = true)
 |-- handler_apron: string (nullable = true)
 |-- occupancy_rate: double (nullable = false)

解决方案 #

1. 修复重复列（核心解决步骤） #

从schema中可明确看到存在重复的aibt列，PySpark允许DataFrame存在重复列名，但Pandas不支持该特性，这是触发转换错误的根本原因。

步骤1：定位重复列

from collections import defaultdict

col_counts = defaultdict(int)
for col in flights.columns:
    col_counts[col] += 1
duplicate_cols = [col for col, cnt in col_counts.items() if cnt > 1]
print("重复列：", duplicate_cols)

步骤2：重命名重复列

给重复列添加后缀以区分，再执行转换：

from collections import defaultdict

col_counter = defaultdict(int)
new_col_names = []
for col in flights.columns:
    col_counter[col] += 1
    if col_counter[col] > 1:
        new_col_names.append(f"{col}_{col_counter[col]}")
    else:
        new_col_names.append(col)

# 重命名DataFrame的列
flights_clean = flights.toDF(*new_col_names)

# 执行转换
pd_flights = flights_clean.toPandas()

2. 确保Arrow优化配置生效（可选） #

若关闭Arrow优化后配置未生效，可尝试重启Databricks集群后重新设置：

spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "false")
spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")

3. 分批转换（大数据量场景适配） #

如果DataFrame数据量过大，修复重复列后仍出现转换问题，可尝试分批转换后合并：

import pandas as pd

# 设置每批数据量
batch_size = 10000
total_rows = flights_clean.count()
# 计算分批数量
num_batches = (total_rows // batch_size) + 1
# 拆分DataFrame并分批转换
batches = flights_clean.randomSplit([1]*num_batches)
pd_flights = pd.concat([batch.toPandas() for batch in batches])

内容的提问来源于stack exchange，提问作者Hans.nl