我正在使用pandas和Spark数据框架。数据框架总是非常大(>20 GB),标准的Spark函数对于这些大小不足。目前,我将我的pandas数据帧转换为Spark数据帧,如下所示:
dataframe = spark.createDataFrame(pandas_dataframe)
我进行这种转换是因为使用Spark将数据框写入HDFS非常容易:
dataframe.write.parquet(output_uri, mode="overwrite", compression="snappy")
但是对于超过2GB的数据框,转换会失败。 如果我将Spark DataFrame转换为Pandas,则可以使用pyarrow:
// temporary write spark dataframe to hdfs
dataframe.write.parquet(path, mode="overwrite", compression="snappy")
// open hdfs connection using pyarrow (pa)
hdfs = pa.hdfs.connect("default", 0)
// read parquet (pyarrow.parquet (pq))
parquet = pq.ParquetDataset(path_hdfs, filesystem=hdfs)
table = parquet.read(nthreads=4)
// transform table to pandas
pandas = table.to_pandas(nthreads=4)
// delete temp files
hdfs.delete(path, recursive=True)
这是从spark到pandas的快速转换,也适用于大于2 GB的数据帧。我还没有找到另一种方法来实现相反的过程,即将pandas数据帧转换为spark,需要使用pyarrow。问题在于我真的找不到如何将pandas数据帧写入到hdfs。
我的pandas版本:0.19.0