apache / apache/beam

Add error context to the BigQuery FAILED_ROWS tagged output

Aperta
#20,522 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub
gcp io new feature P3 python
Lingua principale
Java
Stelle
8.7k
Fork
4.7k
Merge medio
2g 9h
PR unite (30g)
202

Descrizione

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.

Guida per i contributori

Apri la guida per i contributori

Direzione di ricerca

Inizia in sdks/python/apache_beam/io/gcp/bigquery.py, nel percorso di inserimento WriteToBigQuery indicato; esamina come viene assemblato FAILED_ROWS e come viene rappresentato InsertErrorsValueListEntry. Il lavoro è completato quando ogni output non riuscito include la destinazione, i dati della riga e il contesto dell’errore di inserimento.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
google-cloud, python
Ambito
data-engineering
Tipo di issue
Funzionalità
Difficoltà
2/5
Tempo stimato
1-3 ore
Stato di attività
Ferma
Chiarezza
Specificata chiaramente
Idoneità per principianti
55/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.