apache / apache/beam

Add error context to the BigQuery FAILED_ROWS tagged output

Open
#20,522 0 comments 0 reactions 0 assignees View on GitHub
gcp io new feature P3 python
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

When using WriteToBigQuery in the Python SDK, a tagged output of FAILED_ROWS is returned with a list of tuples, each tuple having two elements: a string with the table destination and a dict with the row key value pairs.

This tuple does not, however, contain any error context for why the error failed. I propose adding a third value to the tuple which contains an instance of InsertErrorsValueListEntry to easily provide context for the error in question.

Below is a patch to implement this change (from [https://github.com/apache/beam/blob/e39294dfcab25e2fab250a4691c8ee3ac390976d/sdks/python/apache_beam/io/gcp/bigquery.py#L1186)](https://github.com/apache/beam/blob/e39294dfcab25e2fab250a4691c8ee3ac390976d/sdks/python/apache_beam/io/gcp/bigquery.py#L1186)) - I've made a patch locally and can create a PR if it helps (just need access)

 
```

while True:
passed, errors = self.bigquery_wrapper.insert_rows(
project_id=table_reference.projectId,

dataset_id=table_reference.datasetId,
table_id=table_reference.tableId,
rows=rows,

insert_ids=insert_ids,
skip_invalid_rows=True)

failed_rows = [(rows[entry.index], entry)
for entry in errors]
should_retry = any(
bigquery_tools.RetryStrategy.should_retry(

self._retry_strategy, entry.errors[0].reason) for entry in errors)
if not passed:
message
= (
'There were errors inserting to BigQuery. Will{} retry. '
'Errors were {}'.format((""
if should_retry else " not"), errors))
if should_retry:
_LOGGER.warning(message)
else:

_LOGGER.error(message)

rows = [rows[entry.index] for entry in errors]

if not should_retry:

break
else:
retry_backoff = next(self._backoff_calculator)
_LOGGER.info(
'Sleeping
%s seconds before retrying insertion.', retry_backoff)
time.sleep(retry_backoff)

self._total_buffered_rows
-= len(self._rows_buffer[destination])
del self._rows_buffer[destination]

return [
pvalue.TaggedOutput(

BigQueryWriteFn.FAILED_ROWS,
GlobalWindows.windowed_value((destination, row[0], row[1])))

for row in failed_rows
]

```

Imported from Jira [BEAM-10585](https://issues.apache.org/jira/browse/BEAM-10585). Original Jira may contain additional context.
Reported by: nfbuckley.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.