使用熊猫的“大数据”工作流

Question

在学习熊猫的过程中，我已经尝试了好几个月来找出这个问题的答案。我在日常工作中使用SAS，这是非常好的，因为它提供了非核心支持。然而，SAS作为一个软件是可怕的，原因还有很多。

有一天，我希望用python和pandas取代SAS的使用，但我目前缺乏大型数据集的核心外工作流。我说的不是需要分布式网络的“大数据”，而是文件太大而无法放入内存，但又太小而无法装入硬盘。

我的第一个想法是使用HDFStore将大型数据集保存在磁盘上，并且只将我需要的部分拉入数据帧进行分析。其他人提到MongoDB是一个更容易使用的替代方案。我的问题是：

实现以下目标的最佳实践工作流有哪些：

1. 将平面文件加载到永久的磁盘数据库结构中
2. 查询该数据库以检索数据以输入熊猫数据结构
3. 在操作熊猫的碎片后更新数据库

现实世界中的例子将非常受欢迎，尤其是那些在“大数据”中使用熊猫的人。

编辑--我希望它如何工作的一个示例：

1. 迭代导入大型平面文件并将其存储在永久的磁盘数据库结构中。这些文件通常太大，无法放入内存。
2. 为了使用Pandas，我想读取这些数据的子集（通常一次只读取几列），这些子集可以放入内存。
3. 我将通过对所选列执行各种操作来创建新列。
4. 然后，我必须将这些新列附加到数据库结构中

我正在努力寻找执行这些步骤的最佳实践方法。阅读关于熊猫和pytable的链接，似乎添加一个新的专栏可能是个问题。

编辑——专门回答Jeff的问题：

1. 我正在建立消费者信贷风险模型。数据类型包括电话、SSN和地址特征；财产价值；犯罪记录、破产等贬义信息。。。我每天使用的数据集平均有1000到2000个混合数据类型的字段：数字和字符数据的连续、标称和顺序变量。我很少追加行，但我会执行许多创建新列的操作。

我很少向数据集中添加行。我几乎总是在创建新的专栏（统计/机器学习术语中的变量或特性）。

Answer 1

问题提出两年后，现在出现了一个“核心外”的熊猫等价物：达斯克。太棒了！虽然它不支持所有的熊猫功能，但你可以用它走得更远。更新：在过去两年中，它一直保持不变，并且有大量用户社区与Dask合作。

而现在，在问题提出四年后，Vaex中又出现了一个高性能的“核心外”熊猫。它“使用内存映射、零内存复制策略和惰性计算以获得最佳性能（无内存浪费）。”它可以处理数十亿行的数据集，而不将它们存储到内存中（甚至可以在次优硬件上进行分析）。

Answer 2

我认为上面的答案缺少了一个我发现非常有用的简单方法。

当我有一个文件太大而无法加载到内存中时，我会将该文件分解为多个较小的文件（按行或列）

例如：如果30天的交易数据价值约为30GB大小，我将其分解为每天约1GB大小的文件。我随后分别处理每个文件，并在最后汇总结果

最大的优点之一是它允许并行处理文件（多线程或进程）

另一个优点是，文件操作（如示例中的添加/删除日期）可以通过常规shell命令完成，这在更高级/复杂的文件格式中是不可能的

这种方法并不能涵盖所有的场景，但是在很多场景中非常有用

Answer 3

我经常以这种方式使用数十GB的数据，例如，我在磁盘上有表，通过查询读取、创建数据并追加回来。

关于如何存储数据的一些建议，值得阅读文档和本线程的后期。

会影响你如何存储数据的细节，比如：
尽可能多地提供细节；我可以帮助你开发一个结构。

1. 数据的大小，行，列，列的类型；你是追加行，还是只是列？
2. 典型的操作会是什么样子。例如，对列进行查询，以选择一组行和特定列，然后进行操作（内存中），创建新列，保存这些列。
   （举一个玩具例子可以让我们提供更具体的建议。）
3. 处理完之后，你会怎么做？第2步是特别的，还是可重复的？
4. 输入平面文件：有多少，粗略的总大小，以Gb为单位。这些是如何组织的，例如通过记录？是每个文件都包含不同的字段，还是每个文件都有一些记录，每个文件中包含所有字段？
5. 您是否曾经根据标准选择行（记录）的子集（例如，选择具有字段A的行

确保至少安装了0.10.1的熊猫。

逐块读取迭代文件和多个表查询。

由于pytable被优化为按行操作（这是您查询的内容），我们将为每组字段创建一个表。这样可以很容易地选择一组字段（这将与一个大表一起工作，但这样做更有效...我想我将来可能能够修复这个限制...不管怎样，这是更直观的）：
（以下是伪代码。）

import numpy as np
import pandas as pd

# create a store
store = pd.HDFStore('mystore.h5')

# this is the key to your storage:
#    this maps your fields to a specific group, and defines 
#    what you want to have as data_columns.
#    you might want to create a nice class wrapping this
#    (as you will want to have this map and its inversion)  
group_map = dict(
    A = dict(fields = ['field_1','field_2',.....], dc = ['field_1',....,'field_5']),
    B = dict(fields = ['field_10',......        ], dc = ['field_10']),
    .....
    REPORTING_ONLY = dict(fields = ['field_1000','field_1001',...], dc = []),

)

group_map_inverted = dict()
for g, v in group_map.items():
    group_map_inverted.update(dict([ (f,g) for f in v['fields'] ]))

读取文件并创建存储（基本上做append\u to\u multiple所做的事情）：

for f in files:
   # read in the file, additional options may be necessary here
   # the chunksize is not strictly necessary, you may be able to slurp each 
   # file into memory in which case just eliminate this part of the loop 
   # (you can also change chunksize if necessary)
   for chunk in pd.read_table(f, chunksize=50000):
       # we are going to append to each table by group
       # we are not going to create indexes at this time
       # but we *ARE* going to create (some) data_columns

       # figure out the field groupings
       for g, v in group_map.items():
             # create the frame for this group
             frame = chunk.reindex(columns = v['fields'], copy = False)    

             # append it
             store.append(g, frame, index=False, data_columns = v['dc'])

现在，您在文件中拥有了所有的表（实际上，如果您愿意，您可以将它们存储在单独的文件中，您可能必须将文件名添加到group_map，但这可能不是必要的）。

以下是获取列和创建新列的方式：

frame = store.select(group_that_I_want)
# you can optionally specify:
# columns = a list of the columns IN THAT GROUP (if you wanted to
#     select only say 3 out of the 20 columns in this sub-table)
# and a where clause if you want a subset of the rows

# do calculations on this frame
new_frame = cool_function_on_frame(frame)

# to 'add columns', create a new group (you probably want to
# limit the columns in this new_group to be only NEW ones
# (e.g. so you don't overlap from the other tables)
# add this info to the group_map
store.append(new_group, new_frame.reindex(columns = new_columns_created, copy = False), data_columns = new_columns_created)

当你准备好post_processing：

# This may be a bit tricky; and depends what you are actually doing.
# I may need to modify this function to be a bit more general:
report_data = store.select_as_multiple([groups_1,groups_2,.....], where =['field_1>0', 'field_1000=foo'], selector = group_1)

关于数据列，实际上不需要定义任何数据列；它们允许您根据列对子选择行。例如，类似于：

store.select(group, where = ['field_1000=foo', 'field_1001>0'])

在最终报告生成阶段，您可能对它们最感兴趣（基本上，一个数据列与其他列是分离的，如果您定义了很多，这可能会在一定程度上影响效率）。

您可能还想：

• 创建一个函数，该函数获取字段列表，在groups\u映射中查找组，然后选择这些组并连接结果，从而获得结果帧（这基本上就是select\u as\u multiple所做的）。这样，结构对您来说就相当透明了。
• 某些数据列上的索引（使行子集设置更快）。
• 启用压缩

让我知道，当你有问题！