如何在 Jupyter notebook 中构建大量数据的机器学习模型？答案

【问题标题】：How to build machine learning model in Jupyter notebook with large amount of data?如何在 Jupyter notebook 中构建大量数据的机器学习模型？
【发布时间】：2019-12-20 17:46:50
【问题描述】：

背景：

我在 Google Cloud BigQuery 中有大量数据 (1500GB)。

我正在尝试使用这些数据作为训练数据集来构建 ML 模型。所以我在 Jupyter notebook 中编写了以下代码来获取数据集。

import pandas as pd
from google.cloud import bigquery

import os
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = './my_credential.json'

client = bigquery.Client()

sql = """
    SELECT
Feature1,
Feature2,
Feature3,
target
FROM dataset
    """

sql_result = client.query(
    sql
)
sql_result.to_dataframe()

问题：

代码在执行 30 分钟后抛出内存错误。我知道这是因为代码试图将 1500GB 数据提取到我的 Jupyter 笔记本，但我不知道如何修复。

如何使用 Jupyter notebook 对如此大量的数据进行训练？

【问题讨论】：

您是否考虑过发送多个查询以仅检索 N 个样本而不是查询整个数据集，其中 N 是您在训练期间的批量大小？您也可以进行随机查询 javatpoint.com/sql-select-random 但它们必须是非- 重叠
@JacoSolari 感谢您的建议。这是个好主意。但是我怎么知道我已经覆盖了所有 1500GB 数据集？是否存在对相同数据进行训练的风险（因为样本是随机的）？
是的，这就是我所说的不重叠。我不是 SQL 专家，但我猜你的每个样本都有一个唯一的 ID。您可以在每个随机查询中存储唯一 ID（假设您填写了一个列表）并从下一个查询中排除那些“已查询”的 ID。

标签： python machine-learning google-bigquery jupyter-notebook bigdata

【解决方案1】：

核外计算引擎可能正是您想要的。由于您使用的是 Python，因此您应该查看 Dask 和 Apache Spark 和 PySpark。

Dask 是一个用 Python 实现的轻量级库，它位于 NumPy 和 pandas 之上，允许并行和/或核外计算。它通过dask-ml 集成了机器学习，并公开了与 NumPy、pandas 和 Scikit-Learn 的 API 非常相似的 API。

Spark 是一个在 Scala 中实现的包罗万象的框架，在 JVM 上运行并为 Python 公开了一个 API。它在大数据处理的行业中更加成熟和广泛使用。 Spark 还提供了一个机器学习库MLLib。

两者都可以在您的本地机器或专用多节点集群上运行，以加快计算速度。

【讨论】：

【解决方案2】：

您打算为该文件使用多少内存？ 1500GB 是一个数字，请注意，这不是 python 用作内存的最终大小，这里是关于 python 内存使用的起点：Why do ints require three times as much memory in Python?

我将按以下步骤进行：

你真的需要总数据作为训练集吗？或者你能减少它吗
您需要检查incremental learning，这是一个逐步学习的概念。

这里有一些进一步的解释：https://datascience.stackexchange.com/questions/27767/opening-a-20gb-file-for-analysis-with-pandas

【讨论】：

【解决方案3】：

感谢 JacoSolari。我通过逐块读取 1500GB 数据找到了解决方法。

请参阅以下代码以供参考。程序每次读取8000行，第一次读取0~7999行，第二次读取8000~15999行，...

# read data by chunk
class DataChunkReader():
    def __init__(self):
        dataset = client.dataset('dataset_name', project='project_name')
        table_ref = dataset.table('table_name')
        self.table = client.get_table(table_ref)
        self.start_index = 0
        self.max_results = 8000 # read 8000 rows as a chunk every time

    def read_a_chunk(self):
        rows = client.list_rows(self.table, start_index = self.start_index, max_results = self.max_results).to_dataframe()
        self.start_index += self.max_results
        return rows

    def reset(self):
        self.start_index = 0

data_reader = DataChunkReader()

for i in range(2):
    df = data_reader.read_a_chunk()
    print(df)

【讨论】：