来自 spark 数据帧的块 topandas

Question

我有一个包含 1000 万条记录和 150 列的 spark 数据框。我正在尝试将其转换为熊猫 DF。

x = df.toPandas()
# do some things to x

它失败了ordinal must be >= 1。我假设这是因为一次处理太大了。是否可以将其分块并将其转换为每个块的熊猫 DF？

全栈：

ValueError                                Traceback (most recent call last)
<command-2054265283599157> in <module>()
    158 from db.table where snapshot_year_month=201806""")
--> 159 ps = x.toPandas()
    160 # ps[["pol_nbr",
    161 # "pol_eff_dt",

/databricks/spark/python/pyspark/sql/dataframe.py in toPandas(self)
   2029                 raise RuntimeError("%s\n%s" % (_exception_message(e), msg))
   2030         else:
-> 2031             pdf = pd.DataFrame.from_records(self.collect(), columns=self.columns)
   2032 
   2033             dtype = {}

/databricks/spark/python/pyspark/sql/dataframe.py in collect(self)
    480         with SCCallSiteSync(self._sc) as css:
    481             port = self._jdf.collectToPython()
--> 482         return list(_load_from_socket(port, BatchedSerializer(PickleSerializer())))
    483

Answer 1

如果您的表有一个整数键/索引，您可以使用循环+查询来读取大数据帧的块。

我远离df.toPandas()，它会带来很多开销。相反，我有一个辅助函数，它将查询结果（实例pyspark列表）转换为.Rowpandas.DataFrame

In [1]: from pyspark.sql.functions import col

In [2]: from pyspark.sql import SparkSession

In [3]: import numpy as np

In [4]: import pandas as pd

In [5]: def to_pandas(rows):
       :     row_dicts = [r.asDict() for r in rows]
       :     return pd.DataFrame.from_dict(row_dicts)
       :

要查看此函数的实际效果，让我们制作一个小型示例数据框。

In [6]: from string import ascii_letters
       : n = len(ascii_letters)
       : df = pd.DataFrame({'id': range(n),
       :                    'num': np.random.normal(10,1,n),
       :                    'txt': list(ascii_letters)})
       : df.head()
Out [7]:
   id        num txt
0   0   9.712229   a
1   1  10.281259   b
2   2   8.342029   c
3   3  11.115702   d
4   4  11.306763   e


In [ 8]: spark = SparkSession.builder.appName('Ops').getOrCreate()
       : df_spark = spark.createDataFrame(df)
       : df_spark
Out[ 9]: DataFrame[id: bigint, num: double, txt: string]

通过索引过滤来收集块。

In [10]: chunksize = 25
       : for i in range(0, n, chunksize):
       :     chunk = (df_spark.
       :               where(col('id').between(i, i + chunksize)).
       :               collect())
       :     pd_df = to_pandas(chunk)
       :     print(pd_df.num.mean())
       :
9.779573360741152
10.23157424753804
9.550750629366462