我在databricks笔记本上有一个很大的数据框(2000万行,35列)。我使用python对其执行了一些转换和连接(合并)操作,例如:
mdf.path_info = mdf.path_info.transform(modify_path_info)
x = mdf[['providerid','domain_name']].groupby(['providerid']).apply(domain_features)
mdf = ks.merge( mdf, x[['domain_namex','domain_name_grouped']], left_index=True, right_index=True)
x = mdf.groupby(['providerid','uid']).apply(userspecificdetails)
mmdf = mdf.merge(x[['providerid','uid',"date_last_purch","lifetime_value","age"]], how="left", on=['providerid','uid'])
在这些操作之后,我想显示dataframe的一些行来验证结果dataframe。我正在尝试打印/显示这个大Dataframe的1-5行,但是由于spark的惰性计算特性,所有的print命令都会启动6-12个spark作业并永远运行,之后集群将进入不可用状态,然后什么都不会发生。
mdf.head()
display(mdf)
mdf.take([1])
mdf.iloc[0]
还尝试转换为sparkDataframe,然后尝试:
df = mdf.to_spark()
df.show(1)
df.rdd.takeSample(False, 1, seed=0)
df.first()
我使用的集群配置是8worker\u4core\u8gb,这意味着在databricks运行时版本7.0(包括apachespark3.0.0和scala 2.12)上,每个worker和driver节点都是8.0gb内存、4核、0.5dbu
有人能不能建议一种更快、更快速的方法来获取/打印一行大Dataframe,而不是等待处理整个2000万行Dataframe。
2条答案
按热度按时间cwdobuhd1#
由于延迟求值,所以在编写时,spark将首先执行转换,然后显示一行。您可以做的是减小输入数据的大小,并在更小的数据集上进行转换,例如:
https://spark.apache.org/docs/3.0.0/api/python/pyspark.sql.html#pyspark.sql.dataframe.sample
cu6pst1q2#
你可以的
cache
转换为spark dataframe后的计算结果,然后调用action
.您可能需要释放用于缓存的内存: