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