我有一份工作需要在分区的spark dataframe上运行,并且该过程如下:
rdd = sp_df.repartition(n_partitions, partition_key).rdd.mapPartitions(lambda x: some_function(x))
结果是一个含有
pandas.dataframe
的rdd
。type(rdd) => pyspark.rdd.PipelinedRDD
type(rdd.collect()[0]) => pandas.core.frame.DataFrame
rdd.glom().collect()
的返回结果如下:
[[df1], [df2], ...]
现在我希望将结果转换为Spark数据框,我的做法是:
sp = None
for i, partition in enumerate(rdd.collect()):
if i == 0:
sp = spark.createDataFrame(partition)
else:
sp = sp.union(spark.createDataFrame(partition))
return sp
然而,结果可能会很大,
rdd.collect()
可能会超出驱动程序的内存限制,因此我需要避免使用collect()
操作。有什么方法可以解决这个问题吗?提前感谢!
rdd.toDF()
。或者,spark.createDataFrame(rdd)
。 - samkartValueError: The truth value of a DataFrame is ambiguous. Use a.empty, a.bool(), a.item(), a.any() or a.all().
,我猜它只适用于Row
。 - MJeremycreateDataFrame(x, schema="uuid_id STRING, url STRING, title STRING")
我试图在文档中找到它支持哪些类型,但那些基本上是Python数据类型在SQL符号中的类比。 - dre-hh