Out-of-Memory Data Analytics with Python¶
Python and Big Data¶
Python per se is not a Big Data technology. However, Python in combination with libraries like pandas or PyTables allows the management and the analysis of quite large data sets. This IPython Notebook illustrates some Python approaches in this regard.
For our purposes, we define Big Data as a (number of) object(s) and/or data file(s) that do(es) not fit into the memory of a single computer (server, node, etc.) – whatever hardware your are using for data analytics. On such a data file, typical analytics tasks, like counting, aggregation and selection, shall be implemented.
Out-of-Memory Analytics with NumPy¶
Sometimes operations on NumPy ndarray objects generate so many temporary objects that the available memory does not suffice to finish the desired operation. An example might be a.dot(a.T), i.e. the dot product of an array a with iteself transposed.
Such an operation needs memory for three arrays: a, a.T and a.dot(a.T). If the array a is sufficiently large, say 50% of the free memory, such an operation is impossible with the usual approach.
A solution is to work with disk-based arrays and to use memory maps of these arrays.
Some imports first and a check of the free memory.
import numpy as np
import psutil
psutil.phymem_usage()
Sample Data¶
We generate a larger NumPy ndarray object.
m = 13500
n = 13500
%%time
a = np.random.standard_normal((m, n))
a.nbytes
# array of 1.45GB in size
Checking memory again – and that the object (reference pointer) indeed owns the data.
psutil.phymem_usage()
# free memory less than needed
# (multiple) copies/temporaries impossible
a.flags.owndata
# the object owns the in-memory data
Simple operations on the in-memory ndarray object.
a[:3, :3]
# sample data
%time a.mean()
# reductions work
Now save this object to disk ...
path = './data/'
%time np.save(path + 'od', a)
# save memory array to disk (SSD)
# needs less time than in-memory generation
... and delete the in-memory object.
del a
# delete the in-memory version
# to free memory -- somehow ...
# gc does not work "instantly"
psutil.phymem_usage()
# garbage collection does not bring that much ...
# memory usage has not changed significantly
Memory Map of Data¶
Using the saved object, we generate a new memmap object.
od = np.lib.format.open_memmap(path + 'od.npy', dtype=np.float64, mode='r')
# open memmap array with the array file as data
od.flags.owndata
# object does not own the data
It mainly behaves the same way as in-memory ndarray objects behave.
od[:3, :3]
# compare sample data
%time od.mean()
# operations in NumPy as usual
# somewhat slower of course ...
Memory Maps of (Intermediate) Results¶
Major memory problems with NumPy ndarray objects generally arise due to temporary arrays needed to store intermediate results. We therefore generate memmap objects to store intermediate and final results.
First, for the transpose of the array.
tr = np.memmap(path + 'tr.npy', dtype=np.float64, mode='w+', shape=(n, m))
# memmap object for transpose
%time tr[:] = od.T[:]
# write transpose to disk
Second, for the final results.
re = np.memmap(path + 're.npy', dtype=np.float64, mode='w+', shape=(m, m))
# memmap object for result
%time re[:] = od.dot(tr)[:]
# store results on disk
Final Look and Cleaning Up¶
4+ GB of data (od + tr + re) crunched on a notebook with 4 GB of RAM.
ll ./data/
!rm ./data/*
Using a Sub-Process¶
The futures module allows the use of a separate sub-process for callables.
import futures
# the callable
def generate_array_on_disk(m, n):
a = np.random.standard_normal((m, n))
# memory inefficient operation
np.save(path + 'od.npy', a)
The use of such a sub-process makes sure that any memory used by the sub-process get immediately freed after the sub-process is terminated. This leaves the free memory of the current process mainly unchanged. Avoids "unpredictable" behaviour of Python garbage collection.
psutil.phymem_usage()
%%time
with futures.ProcessPoolExecutor(max_workers=1) as subprocess:
subprocess.submit(generate_array_on_disk, m, n).result()
# separate sub-process is started, the callable is executed
# the process with all its memory usage is killed
psutil.phymem_usage()
# meanwhile memory was freed again
Final look and clean-up.
ll ./data/
!rm ./data/*
Processing (Too) Large CSV Files¶
We generate a CSV file on disk that is too large to fit into memory. We process this file with the help of pandas and PyTables.
First, some imports.
import os
import numpy as np
import pandas as pd
import datetime as dt
Generating an Example CSV File¶
Number of rows to be generated for random data set.
N = int(1e7)
N
Using both random integers as well as floats.
ran_int = np.random.randint(0, 10000, size=(2, N))
ran_flo = np.random.standard_normal((2, N))
Path to the files and filename for csv file.
path = './data/'
csv_name = path + 'data.csv'
Writing the data row by row.
%%time
with open(csv_name, 'wb') as csv_file:
header = 'date,int1,int2,flo1,flo2\n'
csv_file.write(header)
for _ in xrange(10):
# 10 times the original data set
for i in xrange(N):
row = '%s, %i, %i, %f, %f\n' % \
(dt.datetime.now(), ran_int[0, i], ran_int[1, i],
ran_flo[0, i], ran_flo[1, i])
csv_file.write(row)
print os.path.getsize(csv_name)
Delete the original NumPy ndarray objects.
del ran_int
del ran_flo
Reading some rows to check the content.
with open(csv_name, 'rb') as csv_file:
for _ in xrange(5):
print csv_file.readline(),
Reading and Writing with pandas¶
The filename for the pandas HDFStore.
pd_name = path + 'data.h5p'
h5 = pd.HDFStore(pd_name, 'w')
pandas allows to read data from (large) files chunk-wise via a file-iterator.
it = pd.read_csv(csv_name, iterator=True, chunksize=N / 20)
Reading and storing the data chunk-wise (this processes roughly 12 GB of data).
%%time
for i, chunk in enumerate(it):
h5.append('data', chunk)
if i % 20 == 0:
print os.path.getsize(pd_name)
The resulting HDF5 file – with 100,000,000 rows and five columns.
h5
Disk-Based Analytics with pandas¶
The disk-based pandas DataFrame mainly behaves like an in-memory object – but these operations are not memory efficient.
# h5['data'].describe()
Data selection and plotting works as with regular pandas DataFrame objects – again not really memory efficient.
# %matplotlib inline
# h5['data']['flo2'][0:N:500].cumsum().plot()
h5.close()
The major reason is that the DataFrame data structure is broken up (e.g. columns) during storage. For analytics it has to be put together in-memory again.
import tables as tb
h5 = tb.open_file('./data/data.h5p', 'r')
h5
h5.close()
Reading with pandas and Writing with PyTables¶
The PyTables database file.
import tables as tb
tb_name = path + 'data.h5t'
h5 = tb.open_file(tb_name, 'w')
Using a rec array object of NumPy to provide the row description for the PyTables table. To this end, a custom dtype object is needed.
dty = np.dtype([('date', 'S26'), ('int1', '<i8'), ('int2', '<i8'),
('flo1', '<f8'), ('flo2', '<f8')])
# change dtype for date from object to string
Adding compression to the mix (less storage, better backups, better data transfer, etc.).
filters = tb.Filters(complevel=5, complib='blosc')
Again reading and writing chunk-wise, this time appending to a PyTables table object (this processes again roughly 12GB of uncompressed data).
it = pd.read_csv(csv_name, iterator=True, chunksize=N / 20)
%%time
tab = h5.create_table('/', 'data', np.array(it.read().to_records(index=False), dty),
filters=filters)
# initialize table object by using first chunk and adjusted dtype
for chunk in it:
tab.append(chunk.to_records(index=False))
tab.flush()
The resulting table object – 100,000,000 rows and five columns.
h5.get_filesize()
tab
Out-of-Memory Analytics with PyTables¶
Data on disk can be used as if it would be both in-memory and uncompressed. De-compression is done at run-time.
tab[N:N + 3]
# slicing row-wise
tab[N:N + 3]['date']
# access selected data points
Counting of rows is easily accomplished (although here not really needed).
%time len(tab[:]['flo1'])
# length of column (object)
Aggregation operations, like summing up or calculating the mean value, are another application area.
%time tab[:]['flo1'].sum()
# sum over column
%time tab[:]['flo1'].mean()
# mean over column
Typical, SQL-like, conditions and queries can be added.
%time sum([row['flo2'] for row in tab.where('(flo1 > 3) & (int2 < 1000)')])
# sum combined with condition
h5.close()
Overview¶
All operations have been on data sets that do not fit (if uncompressed) into the memory of the machine they haven been implemented on.
ll data/*
Using compression of course reduces the size of the PyTables table object relative to the csv and the pandas HDFStore files. This might, in certain circumstances, lead to file sizes that would again fit in memory.
!rm data/*
Conclusions¶
Python and libraries like NumPy, pandas, PyTables provide useful means and approaches to circumvent the limitations of free memory on a single computer (node, server, etc.).
Key to the performance of such out-of-memory operations are mainly the storage hardware (speed/capacity), the data format used (e.g. HDF5 vs. relational databases) and in some scenarios also the use of performant compression algorithms.
Reading writing speed of SSD hardware is evolving fast:
- status quo: 512 MB/s reading/writing (e.g. MacBook)
- available: 1 GB/s reading/writing (e.g. for servers)
- state-of-the-art: 2 & 1 GB/s reading & writing speed (e.g. from Intel)
http://fastestssd.com: "800GB Intel SSD Announced, Offers 2 GB/s Throughputs"
The Python Quants GmbH – the company Web site
Dr. Yves J. Hilpisch – my personal Web site
Python for Finance – my NEW book (out as Early Release)
Derivatives Analytics with Python – my derivatives analytics book
www.derivatives-analytics-with-python.com
Contact Us