`EventData` is not pickeable and can't work with multiprocessing Pool

[jelther]

• Package Name: azure-eventhub
• Package Version: 5.2.0
• Operating System: Ubuntu 20.04
• Python Version: 3.8

Describe the bug
Object EventData is not pickeable, thus it is not possible to process them in a process pool.

To Reproduce
Steps to reproduce the behavior:

from multiprocessing import Pool
from typing import List
from azure.eventhub import EventData

def build_event_data(i):
	return EventData(body=f'{i}')

if __name__ == "__main__":
	with Pool() as p:
		eventdata : List[EventData] = p.map(build_event_data,[(i) for i in range(10)])

Expected behavior
It's expected that eventdata holds a list of all events created in parallel.

Exception Raised

multiprocessing.pool.MaybeEncodingError: Error sending result: '[EventData(body='0', properties={}, offset=None, sequence_number=None, partition_key=None, enqueued_time=None)]'. Reason: 'TypeError('no default __reduce__ due to non-trivial __cinit__')'

[yunhaoling]

hey @jelther , sorry for the late response.

It is somehow expected as the EventData is composed of C object, to make it pickable. it is required to overload the __reduce__ method (due to the non-trivial cinit).

This is a topic on object serialization, the team would still need some time to discuss on different approaches we want for this kind of scenario.

to better understand your use case:

• do you use any properties on the EventData other than body?
  • if that's the case, you might just need to pass EventData.body (which is a generator of bytes) or EventData.body_as_str() around to process your data instead of the whole object.
• do you use any schema for the data?

[jelther]

@yunhaoling ,

got it.

• Yes, we use other properties, as well the field properties from EventData object.
• no, we don't use a specific schema for the data.

Our use case is simple : we have a class that acts as an wrapper around a pandas Dataframe.
Based on several properties of this dataframe, we have a method that builds the eventdata and returns it.

Below you can find a snippet with a sample of what we're trying to achieve.

from typing import List

import pandas as pd
from io import BytesIO
from multiprocessing import Pool

from azure.eventhub import EventData


class MyWrapper:
    mydf: pd.DataFrame

    def __init__(self, mydf):
        self.mydf = mydf

    def build_eventdata(self) -> EventData:
        parquetdata = BytesIO()
        self.mydf.to_parquet(path=parquetdata)

        eventdata = EventData(body=parquetdata.getvalue())

        # add some properties to eventdata
        eventdata.properties["dataframerowamount"] = len(self.mydf.index)

        return eventdata


def build_my_event(obj: MyWrapper):
    return obj.build_eventdata()


if __name__ == "__main__":
    object_list: List[MyWrapper] = []
    for i in range(1, 10):
        object_list.append(MyWrapper(mydf=pd.DataFrame([{"col1": "value1"}])))

    # this works pretty well
    my_events: List[EventData] = []
    for obj in object_list:
        my_events.append(obj.build_eventdata())

    # this doesn't
    my_events: List[EventData] = []
    with Pool() as _pool:
        my_events = _pool.starmap(build_my_event, [(obj,) for obj in object_list])