pyspark 为什么Pandas不好?

o7jaxewo  于 2023-03-01  发布在  Spark
关注(0)|答案(1)|浏览(202)

根据文档,PySpark的toPandas方法并不适用于大型数据集,即使启用了Arrow也是如此。
这个方法应该只在生成的Pandas panda.DataFrame应该很小的时候使用,因为所有的数据都被加载到驱动程序的内存中。
尽管如此,我们中的许多人使用它来传输相对较大的 Dataframe (我个人传输过一次5Gb)。
1.为什么toPandas实际上对大数据不好?
1.在什么阈值之后,可以认为 Dataframe 小到足以通过toPandas传输
1.将 Dataframe 传输到本地python最合适的方法是什么?先传输df.repartition(1).write.csv,然后传输hdfs dfs -get
我认为数据集可以很好地适应驱动程序的节点内存

uyhoqukh

uyhoqukh1#

1.因为它会占用内存中的所有数据。如果内存快满了,数据会被卸载到硬盘上的交换内存中,这会大大降低操作速度。特别是当CPU需要访问之前移动到交换内存中的数据时,因为它需要将其放回RAM中,将其他内容移动到交换内存中。如果RAM和交换限制都将超过,您将面临内存不足错误,操作将取消。
1.阈值实际上取决于主机上的可用RAM,但通常建议保持在可能的限制以下,以确保服务器在高负载下的快速操作。
1.一般来说,最好对大数据进行流式处理以避免此问题。通常,您无法知道主机上有多少RAM和交换空间可用。流式处理对大数据来说更安全、更快速。This article有四个优秀的策略,我想在此引用其中两个:
采样:最简单的方法是对数据集进行采样,这种方法在探索阶段特别有效:数据看起来是什么样的?2我可以创建哪些特征?3换句话说,哪些是可行的,哪些是不可行的。4通常,从这样一个大数据集中随机抽取10%的样本就已经包含了大量的信息。5这就提出了第一个问题,你真的需要处理整个数据集来训练一个合适的模型吗?

import pandas
import random

filename = "data.csv" 
n = sum(1 for line in open(filename))-1  # Calculate number of rows > in file
s = n//10  # sample size of 10%
skip = sorted(random.sample(range(1, n+1), n-s))  # n+1 to compensate for header 
df = pandas.read_csv(filename, skiprows=skip)

分块:如果你确实需要处理所有数据,你可以选择将数据分成许多块(这些块本身就适合内存),然后在每个块上执行数据清理和特征工程。此外,根据你想要使用的模型类型,你有两个选择:

  • 如果您选择的模型允许部分拟合,您可以在每个块的数据上递增地训练模型;
  • 在每个数据块上训练一个模型,然后,为了给新的看不见的数据打分,用每个模型做一个预测,取平均值或多数票作为最终预测。
import pandas
from sklearn.linear_model import LogisticRegression
datafile = "data.csv"
chunksize = 100000
models = []
for chunk in pd.read_csv(datafile, chunksize=chunksize):
    chunk = pre_process_and_feature_engineer(chunk) 
    # A function to clean my data and create my features
    model = LogisticRegression()
    model.fit(chunk[features], chunk['label'])
    models.append(model)
df = pd.read_csv("data_to_score.csv")
df = pre_process_and_feature_engineer(df)
predictions = mean([model.predict(df[features]) for model in models], axis=0)

相关问题