Python numpy MemoryError - 将多个 CSV 文件加载到 HDF5 存储中并读入 DataFrame

Question

(使用Python 3.3和Pandas 0.12)

我的问题由两部分组成。

第一

我正在尝试迭代读取/追加多个 csv 文件(总共约 8GB)到基于 this solution 的 HDF5 存储中和 this solution用于创建唯一索引。为什么我开始这样做是因为我读到这样做会产生一个可以快速访问且大小相对较小的文件，从而能够读入内存。然而事实证明我得到了一个 18GB 大的 h5 文件。我的 (Windows) 笔记本电脑有 8GB RAM。我的第一个问题是为什么生成的 h5 比原始 csv 文件的总和大得多？我的第二个问题是为什么我确实没有在表上获得唯一索引？

我的代码如下:

def to_hdf(path):
    """ Function that reads multiple csv files to HDF5 Store """
    # If path exists delete it such that a new instance can be created
    if os.path.exists(path):
        os.remove(path)
    # Creating HDF5 Store
    store = pd.HDFStore(path)

    # Reading csv files from list_files function
    with pd.get_store(path) as store:
        for f in list_files():
            try:
                # Creating reader in chunks -- reduces memory load
                df = pd.read_csv(f, encoding='utf-8', chunksize=50000, index_col=False)
                try:
                    nrows = store.get_storer('ta_store').nrows
                except:
                    nrows = 0
                # Looping over chunks and storing them in store file, node name 'ta_data'
                for chunk in df:
                    # Append chunk to store called 'ta_data'
                    store.append('ta_data', chunk, index=False, min_itemsize={'Placement Ref': 50, 'Click Ref': 50})
            # Print filename if corrupt (i.e. CParserError)
            except (parser.CParserError, ValueError) as detail:
                print(f, detail)

    print("Finished reading to HDF5 store, continuing processing data.")

第二个

我的脚本的第二部分将 HDF5 存储读取到 Pandas DataFrame 中。为什么？因为我需要进行一些数据转换和过滤以获得我想要输出到 csv 文件中的最终数据。但是，任何读取 HDF5 存储的尝试都会收到 MemoryError，使用以下代码:

def read_store(filename, node):
    df = pd.read_hdf(filename, node)
    # Some data transformation and filtering code below

发生此错误的另一个示例是当我想使用以下函数打印存储以显示索引不是唯一的时:

def print_store(filename, node):
    store = pd.HDFStore(filename)
    print(store.select(node))

我的问题首先是如何克服这个 MemoryError 问题。我猜我需要减小 hdf5 文件的大小，但我对编程/python/pandas 很陌生，所以我很高兴收到任何输入。其次，我想知道将存储读入 Pandas DataFrame 是否是进行数据转换(创建一个新列)和过滤(基于字符串和日期时间值)的最有效方法。

非常感谢任何帮助!谢谢:)

编辑

根据要求，来自 csv 文件的审查样本(首先)和 ptdump -av 的结果(如下)

csv 示例

A               B   C               D       E           F           G         H                       I                   J       K               L                               M           N       O
4/28/2013 0:00  1   4/25/2013 20:34 View    Anon 2288 optional1   Optional2   Anon | 306742    252.027.323-306742  8.05    10303:41916417  14613669178715620788:10303      Duplicate   Anon  Display
4/28/2013 0:00  2   4/27/2013 13:40 View    Anon 2289 optional1   Optional2   Anon | 306742    252.027.323-306742  8.05    10303:41916417  14613669178715620788:10303      Duplicate   Anon  Display
4/28/2013 0:00  1   4/27/2013 23:41 View    Anon 5791 optional1   Optional2   Anon | 304142    478.323.464-304142  20.66   10304:37464168  14613663710835083509:10305      Duplicate   Anon  Display
4/28/2013 0:00  1   4/27/2013 16:18 View    Anon 4300 optional1   Optional2   Anon | 304142    196.470.934-304142  3.12    10303:41916420  15013670724970033908:291515610  Normal      Anon  Display

ptdump -av

/ (RootGroup) ''
  /._v_attrs (AttributeSet), 4 attributes:
   [CLASS := 'GROUP',
    PYTABLES_FORMAT_VERSION := '2.1',
    TITLE := '',
    VERSION := '1.0']
/ta_data (Group) ''
  /ta_data._v_attrs (AttributeSet), 14 attributes:
   [CLASS := 'GROUP',
    TITLE := '',
    VERSION := '1.0',
    data_columns := ['F', 'G'],
    encoding := 'UTF-8',
    index_cols := [(0, 'index')],
    info := {'index': {}},
    levels := 1,
    nan_rep := 'nan',
    non_index_axes := [(1, ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O'])],
    pandas_type := 'frame_table',
    pandas_version := '0.10.1',
    table_type := 'appendable_frame',
    values_cols := ['values_block_0', 'values_block_1', 'values_block_2', 'F', 'G']]
/ta_data/table (Table(41957511,)) ''
  description := {
  "index": Int64Col(shape=(), dflt=0, pos=0),
  "values_block_0": Float64Col(shape=(1,), dflt=0.0, pos=1),
  "values_block_1": Int64Col(shape=(1,), dflt=0, pos=2),
  "values_block_2": StringCol(itemsize=30, shape=(11,), dflt=b'', pos=3),
  "F": StringCol(itemsize=50, shape=(), dflt=b'', pos=4),
  "G": StringCol(itemsize=50, shape=(), dflt=b'', pos=5)}
  byteorder := 'little'
  chunkshape := (288,)
  /ta_data/table._v_attrs (AttributeSet), 27 attributes:
   [CLASS := 'TABLE',
    G_dtype := 'bytes400',
    G_kind := ['G'],
    FIELD_0_FILL := 0,
    FIELD_0_NAME := 'index',
    FIELD_1_FILL := 0.0,
    FIELD_1_NAME := 'values_block_0',
    FIELD_2_FILL := 0,
    FIELD_2_NAME := 'values_block_1',
    FIELD_3_FILL := b'',
    FIELD_3_NAME := 'values_block_2',
    FIELD_4_FILL := b'',
    FIELD_4_NAME := 'F',
    FIELD_5_FILL := b'',
    FIELD_5_NAME := 'G',
    NROWS := 41957511,
    F_dtype := 'bytes400',
    F_kind := ['F'],
    TITLE := '',
    VERSION := '2.7',
    index_kind := 'integer',
    values_block_0_dtype := 'float64',
    values_block_0_kind := ['J'],
    values_block_1_dtype := 'int64',
    values_block_1_kind := ['B'],
    values_block_2_dtype := 'bytes240',
    values_block_2_kind := ['E', 'O', 'A', 'H', 'C', 'D', 'L', 'N', 'M', 'K', 'I']]

转换和过滤示例

df['NewColumn'] = df['I'].str.split('-').str[0]

mask = df.groupby('NewColumn').E.transform(lambda x: x.nunique() == 1).astype('bool')
df = df[mask]

Answer 1

• 您需要解析 csv 中的日期，尝试添加 parse_dates = ['A','C']当你 read_csv ;如果你这样做df.get_dtype_count()这些应该显示为 datetime64[ns] ，否则为字符串，占用存储空间大，不易使用

• min_itemsize参数指定此字符串列的最小大小(对于“F”、“G”)；这只是为了保证您的字符串不超过此限制；但它使该列的所有行都具有该宽度。如果您可以降低此值，则会减少您的存储空间

• 您没有创建唯一索引；上面的代码缺少一行。添加df.index = Series(df.index) + nrows读完read_csv

• 您需要以 block 的形式迭代 hdf，就像对 csv 文件进行迭代一样；请参阅here ，并查看有关压缩的文档 here

尚不清楚您的过滤实际上要做什么；你能再解释一下吗？您需要彻底了解 HDF 存储的工作原理(例如，您可以附加行，但不能附加列；可能您需要创建一个结果表，在其中附加转换/过滤的 forws)。您还需要了解索引是如何工作的，您需要一种访问这些行的方法(全局唯一可以做到这一点，但根据数据的结构可能不需要)