I'm writing a Dataflow pipeline using Apache beam to add large batches of rows of data to Bigtable.
apache-beam==2.24.0google-cloud-bigtable==2.4.0I have the following method used in my pipeline to create the Bigtable row(s) prior to writing to Bigtable:
class CreateBigtableRow(beam.DoFn):
def __init__(self, settings):
self.column_family = settings["bigtable_column_family"]
super(CreateBigtableRow, self).__init__()
def process(self, usage_data, *args, **kwargs):
row_key = BigTable.generate_row_key(usage_data, key_order)
return [
BigTable.create_row_and_assign_values(
row_key, usage_data, self.column_family
)
]
where `create_row_and_assign_values is defined as:
def create_row_and_assign_values(
cls, key: str, row: dict, column_family: str
) -> DirectRow:
table_row = DirectRow(key.encode())
for key, val in row.items():
if isinstance(val, float):
val = struct.pack(">d", val)
table_row.set_cell(column_family, key.encode(), val)
return table_row
My pipeline is as follows:
with beam.Pipeline(options=pipeline_options) as pipe:
(
pipe
| beam.Create(["/sample_files/*combined*"]) # reads sample csv file
| fileio.MatchAll()
| fileio.ReadMatches()
| beam.FlatMap(
lambda file: csv.DictReader(open(file.metadata.path))
)
| "Transform to Usage dict" >> beam.ParDo(TransformToBigtableData())
| "Create Bigtable Row" >> beam.ParDo(CreateBigtableRow(bigtable_settings))
| WriteToBigTable(
project_id=bigtable_settings["bigtable_project"],
instance_id=bigtable_settings["bigtable_instance"],
table_id=bigtable_settings["bigtable_table"])
)
The problem I'm having is I get the error
_pickle.PicklingError: Can't pickle <class 'Mutation'>: attribute lookup Mutation on __main__ failed [while running 'Create Bigtable Row']
when running the pipeline. I've added steps to manually batch process the records by using the google-cloud-bigtable library's Bigtable Client, but would prefer to use the build-in WriteToBigTable method as it handles everything for me.
Full stack trace:
Traceback (most recent call last):
File "/app/src/ingest/main.py", line 226, in <module>
run(
File "/app/src/ingest/main.py", line 149, in run
(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/pipeline.py", line 596, in __exit__
self.result = self.run()
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/pipeline.py", line 546, in run
return Pipeline.from_runner_api(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/pipeline.py", line 573, in run
return self.runner.run_pipeline(self, self._options)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/direct/direct_runner.py", line 131, in run_pipeline
return runner.run_pipeline(pipeline, options)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 195, in run_pipeline
self._latest_run_result = self.run_via_runner_api(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 206, in run_via_runner_api
return self.run_stages(stage_context, stages)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 384, in run_stages
stage_results = self._run_stage(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 646, in _run_stage
self._run_bundle(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 769, in _run_bundle
result, splits = bundle_manager.process_bundle(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1080, in process_bundle
result_future = self._worker_handler.control_conn.push(process_bundle_req)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/portability/fn_api_runner/worker_handlers.py", line 378, in push
response = self.worker.do_instruction(request)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 597, in do_instruction
return getattr(self, request_type)(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/sdk_worker.py", line 635, in process_bundle
bundle_processor.process_bundle(instruction_id))
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 995, in process_bundle
input_op_by_transform_id[element.transform_id].process_encoded(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py", line 221, in process_encoded
self.output(decoded_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 354, in output
cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 216, in receive
self.consumer.process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 714, in process
delayed_applications = self.dofn_runner.process(o)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1235, in process
self._reraise_augmented(exn)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1316, in _reraise_augmented
raise new_exn.with_traceback(tb)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1233, in process
return self.do_fn_invoker.invoke_process(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 571, in invoke_process
self.output_processor.process_outputs(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/common.py", line 1396, in process_outputs
self.main_receivers.receive(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 215, in receive
self.update_counters_start(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/operations.py", line 179, in update_counters_start
self.opcounter.update_from(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/opcounters.py", line 211, in update_from
self.do_sample(windowed_value)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/runners/worker/opcounters.py", line 250, in do_sample
self.coder_impl.get_estimated_size_and_observables(windowed_value))
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 1371, in get_estimated_size_and_observables
self._value_coder.get_estimated_size_and_observables(
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 358, in get_estimated_size_and_observables
self.encode_to_stream(value, out, nested)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 422, in encode_to_stream
self.fallback_coder_impl.encode_to_stream(value, stream, nested)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/coders/coder_impl.py", line 262, in encode_to_stream
return stream.write(self._encoder(value), nested)
File "/opt/pysetup/.venv/lib/python3.9/site-packages/apache_beam/coders/coders.py", line 800, in <lambda>
lambda x: dumps(x, protocol), pickle.loads)
_pickle.PicklingError: Can't pickle <class 'Mutation'>: attribute lookup Mutation on __main__ failed [while running 'Create Bigtable Row']
Your google-cloud-bigtable version is too high.
There is some movement in updating apache-beam dependencies here
They have the same issue. Can you roll back your bigtable version to something before 2? If you run this:
pip install apache-beam[gcp]
It'll install the recommended version.