Streaming, Serialization, and IPC¶
Writing and Reading Streams¶
Arrow defines two types of binary formats for serializing record batches:
- Streaming format: for sending an arbitrary length sequence of record batches. The format must be processed from start to end, and does not support random access
- File or Random Access format: for serializing a fixed number of record batches. Supports random access, and thus is very useful when used with memory maps
To follow this section, make sure to first read the section on Memory and IO.
Using streams¶
First, let’s create a small record batch:
In [1]: import pyarrow as pa
In [2]: data = [
...: pa.array([1, 2, 3, 4]),
...: pa.array(['foo', 'bar', 'baz', None]),
...: pa.array([True, None, False, True])
...: ]
...:
In [3]: batch = pa.RecordBatch.from_arrays(data, ['f0', 'f1', 'f2'])
In [4]: batch.num_rows
Out[4]: 4
In [5]: batch.num_columns
Out[5]: 3
Now, we can begin writing a stream containing some number of these batches. For
this we use RecordBatchStreamWriter, which can write to a writeable
NativeFile object or a writeable Python object:
In [6]: sink = pa.BufferOutputStream()
In [7]: writer = pa.RecordBatchStreamWriter(sink, batch.schema)
Here we used an in-memory Arrow buffer stream, but this could have been a socket or some other IO sink.
When creating the StreamWriter, we pass the schema, since the schema
(column names and types) must be the same for all of the batches sent in this
particular stream. Now we can do:
In [8]: for i in range(5):
...: writer.write_batch(batch)
...:
In [9]: writer.close()
In [10]: buf = sink.getvalue()
In [11]: buf.size
Out[11]: 1972
Now buf contains the complete stream as an in-memory byte buffer. We can
read such a stream with RecordBatchStreamReader or the
convenience function pyarrow.ipc.open_stream:
In [12]: reader = pa.ipc.open_stream(buf)
In [13]: reader.schema
Out[13]:
f0: int64
f1: string
f2: bool
In [14]: batches = [b for b in reader]
In [15]: len(batches)
Out[15]: 5
We can check the returned batches are the same as the original input:
In [16]: batches[0].equals(batch)
Out[16]: True
An important point is that if the input source supports zero-copy reads
(e.g. like a memory map, or pyarrow.BufferReader), then the returned
batches are also zero-copy and do not allocate any new memory on read.
Writing and Reading Random Access Files¶
The RecordBatchFileWriter has the same API as
RecordBatchStreamWriter:
In [17]: sink = pa.BufferOutputStream()
In [18]: writer = pa.RecordBatchFileWriter(sink, batch.schema)
In [19]: for i in range(10):
....: writer.write_batch(batch)
....:
In [20]: writer.close()
In [21]: buf = sink.getvalue()
In [22]: buf.size
Out[22]: 4210
The difference between RecordBatchFileReader and
RecordBatchStreamReader is that the input source must have a
seek method for random access. The stream reader only requires read
operations. We can also use the pyarrow.ipc.open_file method to open a file:
In [23]: reader = pa.ipc.open_file(buf)
Because we have access to the entire payload, we know the number of record batches in the file, and can read any at random:
In [24]: reader.num_record_batches
Out[24]: 10
In [25]: b = reader.get_batch(3)
In [26]: b.equals(batch)
Out[26]: True
Reading from Stream and File Format for pandas¶
The stream and file reader classes have a special read_pandas method to
simplify reading multiple record batches and converting them to a single
DataFrame output:
In [27]: df = pa.ipc.open_file(buf).read_pandas()
In [28]: df[:5]
Out[28]:
f0 f1 f2
0 1 foo True
1 2 bar None
2 3 baz False
3 4 None True
4 1 foo True
Arbitrary Object Serialization¶
In pyarrow we are able to serialize and deserialize many kinds of Python
objects. While not a complete replacement for the pickle module, these
functions can be significantly faster, particular when dealing with collections
of NumPy arrays.
As an example, consider a dictionary containing NumPy arrays:
In [29]: import numpy as np
In [30]: data = {
....: i: np.random.randn(500, 500)
....: for i in range(100)
....: }
....:
We use the pyarrow.serialize function to convert this data to a byte
buffer:
In [31]: buf = pa.serialize(data).to_buffer()
In [32]: type(buf)
Out[32]: pyarrow.lib.Buffer
In [33]: buf.size
Out[33]: 200028800
pyarrow.serialize creates an intermediate object which can be converted to
a buffer (the to_buffer method) or written directly to an output stream.
pyarrow.deserialize converts a buffer-like object back to the original
Python object:
In [34]: restored_data = pa.deserialize(buf)
In [35]: restored_data[0]
Out[35]:
array([[ 0.28525162, 1.15803552, 1.75955865, ..., 0.06989403,
1.15624031, -0.10755951],
[ 0.18325953, -0.19108787, 0.5361375 , ..., -0.98182569,
-0.28611772, -1.22035724],
[ 0.69519443, -1.41310283, 0.3359842 , ..., -1.15587049,
-2.0310366 , -1.58201493],
...,
[ 2.30785239, -0.02066596, -0.46526787, ..., 0.53222438,
0.7115628 , -0.69658854],
[ 0.08717546, -0.36821839, 0.57433046, ..., -1.17405487,
0.84632265, 0.95738746],
[ 0.51810436, 0.63442531, 0.23127215, ..., -0.2363518 ,
0.81446344, 0.64038853]])
When dealing with NumPy arrays, pyarrow.deserialize can be significantly
faster than pickle because the resulting arrays are zero-copy references
into the input buffer. The larger the arrays, the larger the performance
savings.
Consider this example, we have for pyarrow.deserialize
In [36]: %timeit restored_data = pa.deserialize(buf)
1.61 ms +- 6.05 us per loop (mean +- std. dev. of 7 runs, 1000 loops each)
And for pickle:
In [37]: import pickle
In [38]: pickled = pickle.dumps(data)
In [39]: %timeit unpickled_data = pickle.loads(pickled)
49.6 ms +- 90.4 us per loop (mean +- std. dev. of 7 runs, 10 loops each)
We aspire to make these functions a high-speed alternative to pickle for transient serialization in Python big data applications.
Serializing Custom Data Types¶
If an unrecognized data type is encountered when serializing an object,
pyarrow will fall back on using pickle for converting that type to a
byte string. There may be a more efficient way, though.
Consider a class with two members, one of which is a NumPy array:
class MyData:
def __init__(self, name, data):
self.name = name
self.data = data
We write functions to convert this to and from a dictionary with simpler types:
def _serialize_MyData(val):
return {'name': val.name, 'data': val.data}
def _deserialize_MyData(data):
return MyData(data['name'], data['data']
then, we must register these functions in a SerializationContext so that
MyData can be recognized:
context = pa.SerializationContext()
context.register_type(MyData, 'MyData',
custom_serializer=_serialize_MyData,
custom_deserializer=_deserialize_MyData)
Lastly, we use this context as an additional argument to pyarrow.serialize:
buf = pa.serialize(val, context=context).to_buffer()
restored_val = pa.deserialize(buf, context=context)
The SerializationContext also has convenience methods serialize and
deserialize, so these are equivalent statements:
buf = context.serialize(val).to_buffer()
restored_val = context.deserialize(buf)
Component-based Serialization¶
For serializing Python objects containing some number of NumPy arrays, Arrow
buffers, or other data types, it may be desirable to transport their serialized
representation without having to produce an intermediate copy using the
to_buffer method. To motivate this, suppose we have a list of NumPy arrays:
In [40]: import numpy as np
In [41]: data = [np.random.randn(10, 10) for i in range(5)]
The call pa.serialize(data) does not copy the memory inside each of these
NumPy arrays. This serialized representation can be then decomposed into a
dictionary containing a sequence of pyarrow.Buffer objects containing
metadata for each array and references to the memory inside the arrays. To do
this, use the to_components method:
In [42]: serialized = pa.serialize(data)
In [43]: components = serialized.to_components()
The particular details of the output of to_components are not too
important. The objects in the 'data' field are pyarrow.Buffer objects,
which are zero-copy convertible to Python memoryview objects:
In [44]: memoryview(components['data'][0])
Out[44]: <memory at 0x7fde44117588>
A memoryview can be converted back to a Arrow Buffer with
pyarrow.py_buffer:
In [45]: mv = memoryview(components['data'][0])
In [46]: buf = pa.py_buffer(mv)
An object can be reconstructed from its component-based representation using
deserialize_components:
In [47]: restored_data = pa.deserialize_components(components)
In [48]: restored_data[0]
Out[48]:
array([[-0.71084629, -1.53999613, -0.7271912 , -0.67998188, -0.43663421,
-0.13020289, 0.45710155, 0.79733644, 1.86300956, -1.74382145],
[-1.01786233, 0.46851156, 0.5753886 , -0.01661435, 1.82828141,
-0.93515902, -0.7742813 , -1.41426368, -1.91440675, -1.56477746],
[-0.55694465, -0.78346072, 0.48715291, 0.26055453, -0.25451237,
-0.41015753, -1.44292195, -0.55330622, 0.58754213, 0.20671649],
[ 0.04731158, -1.46171526, -1.5687254 , -0.55467258, -0.12948953,
1.28447203, -0.91136842, -0.7491628 , 0.46239039, 0.94829766],
[ 0.3920792 , -0.61503859, -0.24535991, -1.69108576, -0.07520482,
1.0772701 , -0.75280354, 1.80651946, 0.45267119, 0.30127351],
[-1.81128052, 0.98214069, -1.53875734, 0.76340374, -0.03861162,
-1.2611774 , -0.75656737, 0.3621166 , 0.40285909, -0.83035486],
[-0.49277516, -0.40830261, -0.99326802, 0.96685463, -1.21396892,
-1.16726115, -0.05589027, -1.25347715, -0.01302381, 0.01391371],
[ 0.14893299, 0.12649159, -0.52797847, 1.33158611, -1.50232456,
-0.1038987 , 1.12910026, 2.1465857 , 1.73682528, 0.34759236],
[-0.0499748 , 0.23868989, 0.04434719, -0.339982 , -0.93615099,
-0.33012486, -1.24095936, -2.16599182, -0.29020704, -0.84488452],
[ 2.04802013, 1.29721451, 1.24943045, 1.85607906, -0.04286736,
1.76476354, 0.83804357, 1.71671987, 2.04059127, -0.64489307]])
deserialize_components is also available as a method on
SerializationContext objects.
Serializing pandas Objects¶
The default serialization context has optimized handling of pandas
objects like DataFrame and Series. Combined with component-based
serialization above, this enables zero-copy transport of pandas DataFrame
objects not containing any Python objects:
In [49]: import pandas as pd
In [50]: df = pd.DataFrame({'a': [1, 2, 3, 4, 5]})
In [51]: context = pa.default_serialization_context()
In [52]: serialized_df = context.serialize(df)
In [53]: df_components = serialized_df.to_components()
In [54]: original_df = context.deserialize_components(df_components)
In [55]: original_df
Out[55]:
a
0 1
1 2
2 3
3 4
4 5
Feather Format¶
Feather is a lightweight file-format for data frames that uses the Arrow memory layout for data representation on disk. It was created early in the Arrow project as a proof of concept for fast, language-agnostic data frame storage for Python (pandas) and R.
Compared with Arrow streams and files, Feather has some limitations:
- Only non-nested data types and categorical (dictionary-encoded) types are supported
- Supports only a single batch of rows, where general Arrow streams support an arbitrary number
- Supports limited scalar value types, adequate only for representing typical data found in R and pandas
We would like to continue to innovate in the Feather format, but we must wait for an R implementation for Arrow to mature.
The pyarrow.feather module contains the read and write functions for the
format. The input and output are pandas.DataFrame objects:
import pyarrow.feather as feather
feather.write_feather(df, '/path/to/file')
read_df = feather.read_feather('/path/to/file')
read_feather supports multithreaded reads, and may yield faster performance
on some files:
read_df = feather.read_feather('/path/to/file', nthreads=4)
These functions can read and write with file-like objects. For example:
with open('/path/to/file', 'wb') as f:
feather.write_feather(df, f)
with open('/path/to/file', 'rb') as f:
read_df = feather.read_feather(f)
A file input to read_feather must support seeking.