apache / apache/beam

Add error context to the BigQuery FAILED_ROWS tagged output

Offen
#20,522 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
gcp io new feature P3 python
Vorherrschende Sprache
Java
Sterne
8.7k
Forks
4.7k
Ø Merge
2 T. 9 Std.
Gemergte PRs (30 T.)
202

Beschreibung

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.

Beitragsleitfaden

Beitragsleitfaden öffnen

Rechercherichtung

Beginne in sdks/python/apache_beam/io/gcp/bigquery.py am referenzierten WriteToBigQuery-Einfügepfad; untersuche, wie FAILED_ROWS zusammengestellt wird und wie InsertErrorsValueListEntry dargestellt wird. Fertig ist die Änderung, wenn jede fehlgeschlagene Ausgabe das Ziel, die Zeilendaten und den Kontext des Einfügefehlers enthält.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Bewertung

Tech-Stack
google-cloud, python
Bereich
data-engineering
Issue-Typ
Feature
Schwierigkeit
2/5
Geschätzter Aufwand
1-3 Stunden
Aktivitätsstatus
Veraltet
Klarheit
Klar beschrieben
Anfängerfreundlichkeit
55/100

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.