我的目标是序列化一个RecordBatch
,在websocket通道上发送它 并在接收端对其进行反序列化。
在接收端,接收到数据包并重构后 一个pyarrow.lib.Buffer
对象与pa.py_buffer
,我是 无法将其反序列化回RecordBatch
.
远离 websocket 的样板,这是一个总结我正在尝试做的事情的片段:
import pyarrow as pa
indicators = [(1, 'A'), (2, 'B')]
id = pa.int16()
name = pa.string()
data = pa.array(indicators, type=pa.struct([('id', id), ('name', name)]))
batch = pa.RecordBatch.from_arrays([data], ['indicators'])
buffer = batch.serialize()
# How to get back a RecordBatch from buffer?
#
# ???
当使用这样的serialize
方法时,您可以使用给定已知架构的read_record_batch
函数:
>>> pa.ipc.read_record_batch(buffer, batch.schema)
<pyarrow.lib.RecordBatch at 0x7ff412257278>
但这意味着您需要知道接收端的架构。若要将其封装在序列化数据中,请改用RecordBatchStreamWriter
:
>>> sink = pa.BufferOutputStream()
>>> writer = pa.RecordBatchStreamWriter(sink, batch.schema)
>>> writer.write_batch(batch)
>>> writer.close()
>>> buf = sink.getvalue()
>>> reader = pa.ipc.open_stream(buf)
>>> reader.read_all()
pyarrow.Table
indicators: struct<id: int16, name: string>
child 0, id: int16
child 1, name: string
请参阅 https://arrow.apache.org/docs/python/ipc.html 中的文档