'For' 循环用于将多个 csv 文件从谷歌存储桶读取到 1 个 Pandas DataFrame答案

【问题标题】：'For' loop for reading multiple csv files from a google storage bucket into 1 Pandas DataFrame'For' 循环用于将多个 csv 文件从谷歌存储桶读取到 1 个 Pandas DataFrame
【发布时间】：2020-01-18 20:45:45
【问题描述】：

我目前有 31 个 .csv 文件（都具有相同的结构 - 60 列宽和大约 5000 行深），我正在尝试使用“FOR”循环从谷歌存储桶读取到 1 个熊猫数据帧我在 6 分钟后不断收到“超时”错误。

在进行一些测试时，我注意到我可以一次读取一个 .csv 文件，但是一旦我引入 2 个或更多，我就会收到超时错误。这让我觉得我的代码是问题而不是数据的大小。

代码如下（我应该在 for 循环的任何阶段使用 pd.concat 吗？）将不胜感激

def stage1eposdata(data, context):  

    from google.cloud import storage
    from google.cloud import bigquery
    import pandas as pd
    import dask.dataframe as dd
    import io
    import numpy as np
    import datetime as dt
    from googleapiclient import discovery
    from pandas.io.json import json_normalize
    import google.auth
    import math

    destination_path1 = 'gs://staged_data/ddf-*_stet.csv'  

## Source Buckets #
    raw_epos_bucket = 'raw_data'
    cleaned_epos_bucket = 'staged_data'

    # Confirming Oauth #
    storage_client = storage.Client()
    bigquery_client = bigquery.Client()

    # Confirming Connection #
    raw_epos_data = storage_client.bucket(raw_epos_bucket)
    cleaned_epos_data = storage_client.bucket(cleaned_epos_bucket)

    df  = pd.DataFrame()

    for file in list(raw_epos_data.list_blobs(prefix='2019/')):
        file_path="gs://{}/{}".format(file.bucket.name, file.name)
        df = df.append(pd.read_csv(file_path),sort =False)

    ddf = dd.from_pandas(df,npartitions=1, sort=True)
    ddf.to_csv(destination_path1, index=True, sep=',')

【问题讨论】：

尝试保存在列表中，然后在循环之外使用pd.concat 加入它们。在每个迭代循环中，您将数据帧存储在列表中的某个位置，例如将其称为my_dataframe_list。循环外：pd.concat(my_dataframe_list)。告诉我它是否有效。请注意，这种方式不会在循环的每次迭代中使用 append 方法
那么 my_dataframe_list 代码行在 for 循环中会是什么样子？我很好奇如何避免在此步骤中使用 append 方法
我添加了一个答案，如果它不起作用我尝试编辑它

标签： python-3.x pandas google-cloud-storage dask

【解决方案1】：

试试这个：

## Source Buckets #
    raw_epos_bucket = 'raw_data'
    cleaned_epos_bucket = 'staged_data'

    # Confirming Oauth #
    storage_client = storage.Client()
    bigquery_client = bigquery.Client()

    # Confirming Connection #
    raw_epos_data = storage_client.bucket(raw_epos_bucket)
    cleaned_epos_data = storage_client.bucket(cleaned_epos_bucket)


    my_dataframe_list=[]

    for file in list(raw_epos_data.list_blobs(prefix='2019/')):
        file_path="gs://{}/{}".format(file.bucket.name, file.name)
        my_dataframe_list.append(pd.read_csv(file_path))

    df=pd.concat(my_dataframe_list)
    ddf = dd.from_pandas(df,npartitions=1, sort=True)
    ddf.to_csv(destination_path1, index=True, sep=',')

pd.concat 加入一个 DataFrame 列表。因此，在循环的每次迭代中，您将数据帧保留在列表my_dataframe_list 中，并在循环之外连接列表。如果列匹配，它应该可以工作。

【讨论】：

当我在本地机器上运行它时它可以工作，但是当我将它重构为谷歌云函数时，它只适用于大约一半的 .csv 文件，然后再次超时。我将不得不重新访问谷歌云文档，看看是什么推动了它
您可能已经这样做了，但是 JIC：在这种情况下，您肯定希望增加默认超时。默认情况下为 1 分钟，但您最多可以延长 9 分钟。文档在这里提到了如何使用 CLI 和 GUI：cloud.google.com/functions/docs/concepts/exec#timeout

【解决方案2】：

事实证明，由于它的“惰性”计算特性，dask 可以很好地完成这类事情。我的解决方案如下

## Source Buckets #
raw_epos_bucket = 'raw_data'
cleaned_epos_bucket = 'staged_data'

# Confirming Oauth #
storage_client = storage.Client()
bigquery_client = bigquery.Client()

# Confirming Connection #
raw_epos_data = storage_client.bucket(raw_epos_bucket)
cleaned_epos_data = storage_client.bucket(cleaned_epos_bucket)

my_dataframe_list = []
my_dataframe_list = dd.read_csv('gs://raw_data/*.csv')# '*' is wild card no need to do any more 'For' Loops!

ddf = dd.from_pandas(df,npartitions=1, sort=True)
ddf.to_csv(destination_path1, index=True, sep=',')

【讨论】：