如何将 Google Cloud Storage 中的千兆字节数据加载到 pandas 数据帧中？

Posted 2023-03-24

【中文标题】如何将 Google Cloud Storage 中的千兆字节数据加载到 pandas 数据帧中？

【英文标题】：How do I load gigabytes of data from Google Cloud Storage into a pandas dataframe?

【发布时间】：2017-08-11 18:53:18

【问题描述】：

我正在尝试将千兆字节的数据从 Google Cloud Storage 或 Google BigQuery 加载到 pandas 数据帧中，以便我可以尝试运行 scikit 的 OneClassSVM 和 Isolation Forest（或任何其他一元或 PU 分类）。所以我尝试了pandas-gbq，但尝试运行

pd.read_gbq(query, 'my-super-project', dialect='standard')

导致我的机器在它仅完成 30% 时发出信号。而且我无法在本地加载它，而且我的机器没有足够的空间，而且听起来效率也不高。

我也试过

from google.cloud import storage
client = storage.Client()
bucket = client.get_bucket('bucket-id-here')
# Then do other things...
blob = bucket.get_blob('remote/path/to/file.txt')
print(blob.download_as_string())

我可以加载 1/10 或 1/5 的可用数据，但我的机器最终告诉我它内存不足。

TLDR：有没有一种方法可以让我在云端或某个遥远的超级计算机中运行我的自定义代码（使用 numpy、pandas 甚至 TensorFlow），在那里我可以轻松高效地从 Google 加载数据云存储还是 Google BigQuery？

【问题讨论】：

很遗憾，我找不到read_gbq() 函数，但是如果你在本地下载文件，你可以试试dask.dataframe，它在语法上类似于pandas，但执行的操作是--核心，所以内存应该不是问题。

@DavidDuffrin 我无法下载，因为我的机器没有足够的硬盘空间。

您可以将文件上传到AWS's EMR 并使用PySpark 之类的东西来操作数据吗？我过去曾将 Hadoop 用于类似的“大数据”应用程序。

Cloud Dataflow 是一个选项吗？尝试将数据发送到 AWS 听起来不是一个好的解决方案。

数据有多大？如果它是 20GBs，你可以启动一个有很多内存的 GCE 机器并在那里下载它。如果是 1GB，则需要考虑将整个内容加载到内存中的不同选项（pandas 需要）

【参考方案1】：

我不认为你的方向是正确的。我将尝试解释我通常如何处理数据，希望这能给您一些见解。

我首先倾向于通过应用一些抽样技术或查询更少的天数来处理小型数据集。这一步可以使用pandas或者其他针对小数据开发的工具建模型，提出一些统计，找moments等等。

在我熟悉数据之后，我开始使用大数据工具。

具体来说，我有一个非常小的Dataproc 集群，我已经设置了一个jupyter notebook 来运行pyspark 代码。

集群的总内存必须超过您用作输入的总内存。

使用 pandas 或 spark 数据帧对您来说应该很简单，正如您在 databricks 的 blog post 中看到的那样，spark 已经提供了此功能。

之后，开始实现算法。 Spark 已经提供了一些开箱即用的built-in algorithms，您可以使用它们。

如果您要实现的算法不可用，您可以在他们的存储库中发出请求或自己构建它（您可以使用 Python 的 Scipy 实现作为指导并将其转置到 spark 环境中）。

以下是我如何为我用来为我们公司构建推荐系统的算法之一加载数据的示例：

from pyspark.sql import functions as sfunc
from pyspark.sql import types as stypes

schema = stypes.StructType().add("fv", stypes.StringType()).add("sku", stypes.StringType()).add("score", stypes.FloatType())
train_df = spark.read.csv('gs://bucket_name/pyspark/train_data*.gz', header=True, schema=schema)

Spark 会自动将这些数据分发给集群中可用的不同工作人员。之后我主要运行查询和映射/减少步骤来获取 skus 之间的相关性。

就维护您当前的代码而言，它可能已经无法针对大数据进行扩展。尽管如此，您仍然可以找到很多资源来将 numpy 的强大功能与 spark 结合起来，例如 this example。