使用Pandas提升大HDFStore表的查询性能
我有一个很大的数据表（大约有1.6亿行），我把它存储在硬盘上，像这样：
def fillStore(store, tablename):
files = glob.glob('201312*.csv')
names = ["ts", "c_id", "f_id","resp_id","resp_len", "s_id"]
for f in files:
df = pd.read_csv(f, parse_dates=True, index_col=0, names=names)
store.append(tablename, df, format='table', data_columns=['c_id','f_id'])
这个表有一个时间索引，我会通过c_id和f_id以及时间（通过索引）来查询数据。
我还有另一个数据表，里面有大约18000个“事件”。每个事件包含一些（最少几百条，最多几十万条）单独的记录。我需要为每个事件收集一些简单的统计数据，并将它们存储起来，以便收集一些汇总统计。目前我这样做：
def makeQueryString(c, f, start, stop):
return "c_id == {} & f_id == {} & index >= Timestamp('{}') & index < Timestamp('{}')".format(c, f , str(pd.to_datetime(start)),str(pd.to_datetime(stop)))
def getIncidents(inc_times, store, tablename):
incidents = pd.DataFrame(columns = ['c_id','f_id','resp_id','resp_len','s_id','incident_id'])
for ind, row in inc_times.iterrows():
incidents = incidents.append(store.select(tablename,
makeQueryString(row.c_id,
row.f_id,
row.start,
row.stop))).fillna(ind)
return incidents
这一切都运行得很好，唯一的问题是每个store.select()语句大约需要5秒钟，这意味着处理整整一个月的数据需要24到30个小时。与此同时，我需要的实际统计数据相对简单：
def getIncidentStats(df):
incLen = (df.index[-1]-df.index[0]).total_seconds()
if incLen == 0:
incLen = .1
rqsts = len(df)
rqstRate_s = rqsts/incLen
return pd.Series({'c_id':df.c_id[0],
'f_id':df.fqdn_id[0],
'Length_sec':incLen,
'num_rqsts':rqsts,
'rqst_rate':rqstRate_s,
'avg_resp_size':df.response_len.mean(),
'std_resp_size':df.response_len.std()})
incs = getIncidents(i_times, store, tablename)
inc_groups = incs.groupby('incident_id')
inc_stats = inc_groups.apply(getIncidentStats)
我的问题是：我该如何提高这个工作流程的性能或效率？（请注意，我实际上是将大部分工作批量处理，每次只获取和存储一天的事件，因为我想降低在崩溃时丢失已处理数据的风险。我这里省略了这部分代码，为了简单起见，因为我实际上需要处理整个月的数据。）
有没有办法在我从存储中接收到数据时就开始处理？这样做有什么好处吗？使用store.select_as_index会有帮助吗？如果我接收到一个索引，我仍然需要访问数据来获取正确的统计数据吗？
其他备注/问题：我比较了在SSD和普通硬盘上存储我的HDFStore的性能，发现SSD没有明显的提升。这正常吗？
我还考虑过创建一个大的查询字符串组合，一次性请求所有数据。但当总查询字符串太大（大约5到10个查询）时，会导致内存错误。
编辑 1 如果有关系，我使用的是版本3.1.0的tables和版本0.13.1的pandas。
编辑 2 这里有更多信息：
ptdump -av store.h5
/ (RootGroup) ''
/._v_attrs (AttributeSet), 4 attributes:
[CLASS := 'GROUP',
PYTABLES_FORMAT_VERSION := '2.0',
TITLE := '',
VERSION := '1.0']
/all_recs (Group) ''
/all_recs._v_attrs (AttributeSet), 14 attributes:
[CLASS := 'GROUP',
TITLE := '',
VERSION := '1.0',
data_columns := ['c_id', 'f_id'],
encoding := None,
index_cols := [(0, 'index')],
info := {1: {'type': 'Index', 'names': [None]}, 'index': {'index_name': 'ts'}},
levels := 1,
nan_rep := 'nan',
non_index_axes := [(1, ['c_id', 'f_id', 'resp_id', 'resp_len', 'dns_server_id'])],
pandas_type := 'frame_table',
pandas_version := '0.10.1',
table_type := 'appendable_frame',
values_cols := ['values_block_0', 'c_id', 'f_id']]
/all_recs/table (Table(161738653,)) ''
description := {
"index": Int64Col(shape=(), dflt=0, pos=0),
"values_block_0": Int64Col(shape=(3,), dflt=0, pos=1),
"c_id": Int64Col(shape=(), dflt=0, pos=2),
"f_id": Int64Col(shape=(), dflt=0, pos=3)}
byteorder := 'little'
chunkshape := (5461,)
autoindex := True
colindexes := {
"index": Index(6, medium, shuffle, zlib(1)).is_csi=False,
"f_id": Index(6, medium, shuffle, zlib(1)).is_csi=False,
"c_id": Index(6, medium, shuffle, zlib(1)).is_csi=False}
/all_recs/table._v_attrs (AttributeSet), 19 attributes:
[CLASS := 'TABLE',
FIELD_0_FILL := 0,
FIELD_0_NAME := 'index',
FIELD_1_FILL := 0,
FIELD_1_NAME := 'values_block_0',
FIELD_2_FILL := 0,
FIELD_2_NAME := 'c_id',
FIELD_3_FILL := 0,
FIELD_3_NAME := 'f_id',
NROWS := 161738653,
TITLE := '',
VERSION := '2.6',
client_id_dtype := 'int64',
client_id_kind := ['c_id'],
fqdn_id_dtype := 'int64',
fqdn_id_kind := ['f_id'],
index_kind := 'datetime64',
values_block_0_dtype := 'int64',
values_block_0_kind := ['s_id', 'resp_len', 'resp_id']]
这是主表和inc_times的一些样本：
In [12]: df.head()
Out[12]:
c_id f_id resp_id resp_len \
ts
2013-12-04 08:00:00 637092486 5372764353 30 56767543
2013-12-04 08:00:01 637092486 5399580619 23 61605423
2013-12-04 08:00:04 5456242 5385485460 21 46742687
2013-12-04 08:00:04 5456242 5385485460 21 49909681
2013-12-04 08:00:04 624791800 5373236646 14 70461449
s_id
ts
2013-12-04 08:00:00 1829
2013-12-04 08:00:01 1724
2013-12-04 08:00:04 1679
2013-12-04 08:00:04 1874
2013-12-04 08:00:04 1727
[5 rows x 5 columns]
In [13]: inc_times.head()
Out[13]:
c_id f_id start stop
0 7254 196211 1385880945000000000 1385880960000000000
1 9286 196211 1387259840000000000 1387259850000000000
2 16032 196211 1387743730000000000 1387743735000000000
3 19793 196211 1386208175000000000 1386208200000000000
4 19793 196211 1386211800000000000 1386211810000000000
[5 rows x 4 columns]
关于c_id和f_id，我想从完整存储中选择的ID数量相对较少，和存储中的总ID数量相比。换句话说，在inc_times中，有一些热门ID我会反复查询，而完全忽略一些在完整表中存在的ID。我估计我关心的ID大约占总ID的10%，但这些ID是最受欢迎的，所以它们的记录在整个数据集中占主导地位。
我有16GB的内存。完整存储的大小是7.4G，而完整数据集（作为csv文件）只有8.7GB。起初我以为可以将整个数据加载到内存中，至少进行一些有限的操作，但加载整个数据时出现了内存错误。因此，我将其分批处理为每日文件（完整文件包含一个月的数据）。
1 个回答
4
这里有一些建议，类似的问题可以在这里找到。
使用压缩：可以参考这里。你可以试试这个方法（具体效果可能会让速度变快或变慢，具体情况因人而异）。
ptrepack --chunkshape=auto --propindexes --complevel=9 --complib=blosc in.h5 out.h5
分块使用分层查询。我想说的是，由于你关注的c_id和f_id数量相对较少，可以把单个查询结构化成这样。这有点像使用isin。
f_ids = list_of_f_ids that I care about
c_ids = list_of_c_ids that I care about
def create_batches(l, maxn=32):
""" create a list of batches, maxed at maxn """
batches = []
while(True):
if len(l) <= maxn:
if len(l) > 0:
batches.append(l)
break
batches.append(l[0:maxn])
l = l[maxn:]
return batches
results = []
for f_id_batch in create_batches(f_id_list):
for c_id_batch in create_batches(c_id_list):
q = "f_id={f_id} & c_id={c_id}".format(
f_id=f_id_batch,
c_id=c_id_batch)
# you can include the max/min times in here as well (they would be max/min
# time for ALL the included batches though, maybe easy for you to compute
result = store.select('df',where=q)
# sub process this result
def f(x):
# you will need to filter out the min/max timestamps here (which I gather
# are somewhat dependent on f_id/c_id group
#### process the data and return something
# you could do something like: ``return x.describe()`` for simple stats
results.append(result.groupby(['f_id','c_id').apply(f))
results = pd.concat(results)
这里的关键是处理时，isin的成员数量不能超过32个，这个是numpy/pytables的内部限制。如果超过这个数量，查询会执行，但会丢掉那个变量，并对所有数据进行重新索引（这可不是你想要的结果）。
这样，你在内存中会有一个不错的数据子集，只需几次循环。这些查询的时间大致和你大部分查询差不多，但数量会少很多。
对于给定的子集，查询时间大致是恒定的（除非数据是完全有序的，已经完全索引）。
所以查询会扫描数据的“块”（索引指向的就是这些块）。如果你在很多块中都有很多命中，那么查询速度就会变慢。
这里有个例子
In [5]: N = 100000000
In [6]: df = DataFrame(np.random.randn(N,3),columns=['A','B','C'])
In [7]: df['c_id'] = np.random.randint(0,10,size=N)
In [8]: df['f_id'] = np.random.randint(0,10,size=N)
In [9]: df.index = date_range('20130101',periods=N,freq='s')
In [10]: df.to_hdf('test2.h5','df',mode='w',data_columns=['c_id','f_id'])
In [11]: df.head()
Out[11]:
A B C c_id f_id
2013-01-01 00:00:00 0.037287 1.153534 0.639669 8 7
2013-01-01 00:00:01 1.741046 0.459821 0.194282 8 3
2013-01-01 00:00:02 -2.273919 -0.141789 0.770567 1 1
2013-01-01 00:00:03 0.320879 -0.108426 -1.310302 8 6
2013-01-01 00:00:04 -1.445810 -0.777090 -0.148362 5 5
2013-01-01 00:00:05 1.608211 0.069196 0.025021 3 6
2013-01-01 00:00:06 -0.561690 0.613579 1.071438 8 2
2013-01-01 00:00:07 1.795043 -0.661966 1.210714 0 0
2013-01-01 00:00:08 0.176347 -0.461176 1.624514 3 6
2013-01-01 00:00:09 -1.084537 1.941610 -1.423559 9 1
2013-01-01 00:00:10 -0.101036 0.925010 -0.809951 0 9
2013-01-01 00:00:11 -1.185520 0.968519 2.871983 7 5
2013-01-01 00:00:12 -1.089267 -0.333969 -0.665014 3 6
2013-01-01 00:00:13 0.544427 0.130439 0.423749 5 7
2013-01-01 00:00:14 0.112216 0.404801 -0.061730 5 4
2013-01-01 00:00:15 -1.349838 -0.639435 0.993495 0 9
In [2]: %timeit pd.read_hdf('test2.h5','df',where="f_id=[1] & c_id=[2]")
1 loops, best of 3: 13.9 s per loop
In [3]: %timeit pd.read_hdf('test2.h5','df',where="f_id=[1,2] & c_id=[1,2]")
1 loops, best of 3: 21.2 s per loop
In [4]: %timeit pd.read_hdf('test.2h5','df',where="f_id=[1,2,3] & c_id=[1,2,3]")
1 loops, best of 3: 42.8 s per loop
这个例子是5GB未压缩和2.9GB压缩的数据。这些结果是基于压缩数据的。在这种情况下，使用未压缩的数据实际上会快很多（例如，第一次循环花了3.5秒）。这里有1亿行数据。
所以使用最后一个例子（4），你在查询时间稍微超过3倍的情况下，得到了比第一个例子多9倍的数据。
不过你的加速效果应该会更明显，因为你不会在单独的时间戳上进行选择，而是稍后再处理。
这种方法考虑到了你有足够的主内存来存放批量查询的结果（例如，你在批量查询中选择的是相对较小的一部分数据）。