BigTable: 'Table.mutate_rows' deadline exceeded for large mutations

[cwbeitel]

Deadline exceeded for large mutation of BigQuery table; not necessarily a bug perhaps it would be helpful to other to add this to the docs, automatically batch mutations by size, make the timeout configurable, or document large data transfers as a potential cause of exceeding the deadline.

Environment details

• Cloud BigTable
• Ubuntu 18.04
• Python 3.6.4
• google-cloud-bigtable==0.31.1

Steps to reproduce and code example

Changing the value of mutation_batch_size in the following, too large and the deadline is exceeded; with tfexample serialized video examples with 4 frames of size 224x224. Not necessarily a bug if this is the expected behavior and users should handle this kind of batching themselves.

def iterable_dataset_from_file(filename):
    dataset = tf.data.TFRecordDataset(filename)

    iterator = dataset.make_initializable_iterator()

    next_element = iterator.get_next()

    with tf.Session() as sess:

      sess.run(iterator.initializer)

      i = 0
      while True:
        try:
          if i % 1000 == 0:
            print("Processed %s examples..." % i)
          yield sess.run(next_element)
          i += 1
        except tf.errors.OutOfRangeError:
          print("Ran out of examples (processed %s), exiting..." % i)
          break


def tfrecord_files_to_cbt_table(glob, table, selection, max_records=100000000,
                                mutation_batch_size=250):

  mutation_index = 0

  def new_mutation_batch():
    return [None for _ in range(mutation_batch_size)]

  files = tf.gfile.Glob(glob)

  for file_path in files:
 
    row_mutation_batch = new_mutation_batch()

    for i, example in enumerate(iterable_dataset_from_file(file_path)):

      idx = hashlib.md5(example).hexdigest()
    
      # DEV: To check "shuffle" effect add the id suffix
      idx = "_".join([selection.prefix, idx, str(i)])

      row = table.row(idx)
      row.set_cell(column_family_id=selection.column_family,
                   column=selection.column_qualifier,
                   value=example,
                   timestamp=datetime.datetime.utcnow())

      row_mutation_batch[mutation_index] = row

      if mutation_index == (mutation_batch_size - 1):
        table.mutate_rows(row_mutation_batch)
        row_mutation_batch = new_mutation_batch()
        mutation_index = 0
      else:
        mutation_index += 1

    final_mutation = row_mutation_batch[:(mutation_index-1)]
    if final_mutation:
      table.mutate_rows(final_mutation)

Stack trace

Traceback (most recent call last):
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/grpc_helpers.py", line 79, in next
    return six.next(self._wrapped)
  File "/opt/conda/lib/python3.6/site-packages/grpc/_channel.py", line 341, in __next__
    return self._next()
  File "/opt/conda/lib/python3.6/site-packages/grpc/_channel.py", line 335, in _next
    raise self
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.DEADLINE_EXCEEDED, Deadline Exceeded)>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/opt/conda/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 330, in <module>
    tf.app.run()
  File "/opt/conda/lib/python3.6/site-packages/tensorflow/python/platform/app.py", line 125, in run
    _sys.exit(main(argv))
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 300, in main
    max_records=FLAGS.max_records)
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 228, in tfrecord_files_to_cbt_table
    table.mutate_rows(row_mutation_batch)
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 423, in mutate_rows
    return retryable_mutate_rows(retry=retry)
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 571, in __call__
    mutate_rows()
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/retry.py", line 270, in retry_wrapped_func
    on_error=on_error,
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/retry.py", line 179, in retry_target
    return target()
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 634, in _do_mutate_retryable_rows
    for response in responses:
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/grpc_helpers.py", line 81, in next
    six.raise_from(exceptions.from_grpc_error(exc), exc)
  File "<string>", line 3, in raise_from
google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded

[sduskis]

@crwilcox, @kolea2, @igorbernstein2: this is a problem in the GAPIC config. Currently, the default timeout is set to 60 seconds. Someone may need to increase the timeout for MutateRows calls.

[cwbeitel]

@sduskis Sending data in smaller batches might be a best practice anyway if a part of a large transfer failing could fail the whole transfer (vs. perhaps retries only the failed portion?).

[sduskis]

@cwbeitel, I agree on best practices. In theory, you can use a MutationsBatcher From table.mutations_batcher() which ought to encapsulate the best practices.

I also believe that we ought to review the default timeout value to make sure that it makes sense for this type of RPC.

[tseaver]

Relevant commits:

• c38888d bumped the default timeout for the MutateRows RPC to 600 seconds, which seems like it should be sufficient.
• 3169f10 added a mutation_timeout attribute to Table, and passes it through to the underlying GAPIC method. That value (default None) is passed through to the _RetryableMutateRowsWorker instance used to perform the actual MutateRows RPC call.

An odd side effect of how _RetryableMutateRowsWorker uses the timeout is that it scribbles on the mutate_rows key of the inner_api_calls of the table data client, IFF the wrapper is not already present. So, the first table will have its mutation_timeout value set on the client, but subsequent tables will not. :( A better implementation would probably be just to create the wrapper on the fly each time, if the timeout is not None. Or, even better, just pass through the timeout to the data_client.mutate_rows call, rather than mucking about with wrappers.

[cwbeitel]

@tseaver Ah well done, the ability to configure a timeout seems like a good feature.