-
-
Notifications
You must be signed in to change notification settings - Fork 28
/
message.py
49 lines (36 loc) · 1.15 KB
/
message.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
"""timeflux.core.message: serialize and unserialize dataframes"""
import pickle
import pandas as pd
# import pyarrow as pa
def pickle_serialize(message):
topic = message[0]
data = message[1]
meta = message[2]
return [
topic,
pickle.dumps(data, pickle.HIGHEST_PROTOCOL),
pickle.dumps(meta, pickle.HIGHEST_PROTOCOL),
]
def pickle_deserialize(message):
topic = message[0].decode("utf-8")
data = message[1]
meta = message[2]
return [topic, pickle.loads(data), pickle.loads(meta)]
def msgpack_serialize(message):
# TODO: handle meta and cases where data is None
topic = message[0]
data = message[1]
return [topic, data.to_msgpack()]
def msgpack_deserialize(message):
# TODO: handle meta and cases where data is None
topic = message[0].decode("utf-8")
data = message[1]
return [topic, pd.read_msgpack(data)]
# def arrow_serialize(message):
# topic = message[0].decode('utf-8')
# df = message[1]
# return [topic, pa.serialize(df).to_buffer()]
# def arrow_deserialize(message):
# topic = message[0]
# data = message[1]
# return [topic, pa.deserialize(data)]