如何在 ipython 中将 Spark RDD 转换为 pandas 数据帧?
Posted
技术标签:
【中文标题】如何在 ipython 中将 Spark RDD 转换为 pandas 数据帧?【英文标题】:How to convert Spark RDD to pandas dataframe in ipython? 【发布时间】:2016-01-15 18:34:18 【问题描述】:我有一个RDD
,我想把它转换成pandas
dataframe
。我知道将 RDD
转换为普通的 dataframe
我们可以这样做
df = rdd1.toDF()
但我想将RDD
转换为pandas
dataframe
而不是普通的dataframe
。我该怎么做?
【问题讨论】:
【参考方案1】:你可以使用函数toPandas()
:
以 Pandas pandas.DataFrame 的形式返回此 DataFrame 的内容。
仅当 Pandas 已安装且可用时才可用。
>>> df.toPandas()
age name
0 2 Alice
1 5 Bob
【讨论】:
toDF() 和 toPandas() 有什么区别? @jezrael,如何只转换spark df to pandas df
的前10行?【参考方案2】:
您必须使用 Spark DataFrame 作为 RDD 和所需 Pandas DataFrame 之间的中间步骤。
例如,假设我有一个文本文件 flights.csv
,它已被读入 RDD:
flights = sc.textFile('flights.csv')
你可以检查类型:
type(flights)
<class 'pyspark.rdd.RDD'>
如果你只是在 RDD 上使用toPandas()
,它是行不通的。根据 RDD 中对象的格式,可能需要先进行一些处理才能转到 Spark DataFrame。在这个例子中,这段代码完成了这项工作:
# RDD to Spark DataFrame
sparkDF = flights.map(lambda x: str(x)).map(lambda w: w.split(',')).toDF()
#Spark DataFrame to Pandas DataFrame
pdsDF = sparkDF.toPandas()
你可以检查类型:
type(pdsDF)
<class 'pandas.core.frame.DataFrame'>
【讨论】:
我认为pdsDF = sparkDF.toPandas
缺少 () 来实际调用该方法。应该是:pdsDF = sparkDF.toPandas()
toDF() 和 toPandas() 有什么区别?
toDF() 将 RDD 转换为 Spark DataFrame,toPandas() 将 Spark DataFrame 转换为 Pandas DataFrame。这两种DataFrame是不同的类型。【参考方案3】:
我推荐 joshlk 的快速版 toPandas
import pandas as pd
def _map_to_pandas(rdds):
""" Needs to be here due to pickling issues """
return [pd.DataFrame(list(rdds))]
def toPandas(df, n_partitions=None):
"""
Returns the contents of `df` as a local `pandas.DataFrame` in a speedy fashion. The DataFrame is
repartitioned if `n_partitions` is passed.
:param df: pyspark.sql.DataFrame
:param n_partitions: int or None
:return: pandas.DataFrame
"""
if n_partitions is not None: df = df.repartition(n_partitions)
df_pand = df.rdd.mapPartitions(_map_to_pandas).collect()
df_pand = pd.concat(df_pand)
df_pand.columns = df.columns
return df_pand
<script src="https://gist.github.com/joshlk/871d58e01417478176e7.js"></script>
【讨论】:
以上是关于如何在 ipython 中将 Spark RDD 转换为 pandas 数据帧?的主要内容,如果未能解决你的问题,请参考以下文章
如何在 Spark 中将两个 RDD[string] 合并在一起?
如何在spark(Python)中将两个rdd组合成on rdd