[
https://issues.apache.org/jira/browse/BEAM-10248?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17289188#comment-17289188
]
Chamikara Madhusanka Jayalath commented on BEAM-10248:
------------------------------------------------------
Currently there's not immediate plan to add error handling to FILE_LOADS
method. Also I'm not sure if this is viable (usually, failure from BQ is for
the whole load job not for individual rows). So current recommendation is to
use STREAMING_INSERTS if you need more fine grained error handling. For
FILE_LOADS mode, you can look for error logs in StackDriver logs and if needed
use 'bq' CLI to get more information.
> Beam does not set correct region for BigQuery when requesting load job status
> -----------------------------------------------------------------------------
>
> Key: BEAM-10248
> URL: https://issues.apache.org/jira/browse/BEAM-10248
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
> Affects Versions: 2.22.0
> Environment: -Beam 2.22.0
> -DirectRunner & DataflowRunner
> -Java 11
> Reporter: Graham Polley
> Assignee: Chamikara Madhusanka Jayalath
> Priority: P1
> Fix For: 2.24.0
>
> Attachments: Untitled document.pdf
>
> Time Spent: 50m
> Remaining Estimate: 0h
>
> I am using `FILE_LOADS` (micro-batching) from Pub/Sub and writing to
> BigQuery. My BigQuery dataset is in region `australia-southeast1`.
> If the load job into BigQuery fails for some reason (e.g. table does not
> exist, or schema has changed), then an error from BigQuery is returned for
> the load.
> Beam then enters into a loop of retrying the load job. However, instead of
> using a new job id (where the suffix is incremented by 1), it wrongly tries
> to reinsert the job using the same job id. This is because when it tries to
> look up if the job id already exists, it does not take into consideration the
> region where the dataset is. Instead, it defaults to the US but the job was
> created in `australia-southeast1`, so it returns `null`.
> Bug is on LN308 of `BigQueryHelpers.java`. It should not return `null`. It
> returns `null` because it's looking in the wrong region.
> If I *test* using a dataset in BigQuery that is in the US region, then it
> correctly finds the job id and begins to retry the job with a new job id
> (suffixed with the retry count). We have seen this bug in other areas of Beam
> before and in other tools/services on GCP e.g. Cloud Composer.
> However, that leads me to my next problem/bug. Even if that is fixed, the
> number of retries is set to `Integer.MX_VALUE`, so Beam will keep retrying
> the job 2,147,483,647 times. This is not good.
> The exception is swallowed up by the Beam SDK and never propagated back up
> the stack for users to catch and handle. So, if a load job fails there is no
> way to handle it and react for users.
> I attempted to set `withFailedInsertRetryPolicy()` to transient errors only,
> but it is not supported with `FILE_LOADS`. I also tried, using the
> `WriteResult` object returned from the bigQuery sink/write, to get a handle
> on the error but it does not work. Users need a way to catch and catch failed
> load jobs when using `FILE_LOADS`.
>
> {code:java}
> public class TemplatePipeline {
> private static final String TOPIC =
> "projects/etl-demo-269105/topics/test-micro-batching";
> private static final String BIGQUERY_DESTINATION_FILTERED =
> "etl-demo-269105:etl_demo_fun.micro_batch_test_xxx";
> public static void main(String[] args) throws Exception {
> try {
> PipelineOptionsFactory.register(DataflowPipelineOptions.class);
> DataflowPipelineOptions options = PipelineOptionsFactory
> .fromArgs(args)
> .withoutStrictParsing()
> .as(DataflowPipelineOptions.class);
> Pipeline pipeline = Pipeline.create(options);
> PCollection<PubsubMessage> messages = pipeline
> .apply(PubsubIO.readMessagesWithAttributes().fromTopic(String.format(TOPIC,
> options.getProject())))
> .apply(Window.into(FixedWindows.of(Duration.standardSeconds(1))));
> WriteResult result = messages.apply(ParDo.of(new RowToBQRow()))
> .apply(BigQueryIO.writeTableRows()
> .to(String.format(BIGQUERY_DESTINATION_FILTERED, options.getProject()))
> .withCreateDisposition(CREATE_NEVER)
> .withWriteDisposition(WRITE_APPEND)
> .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
> .withTriggeringFrequency(Duration.standardSeconds(5))
> .withNumFileShards(1)
> .withExtendedErrorInfo()
> .withSchema(getTableSchema()));
> // result.getFailedInsertsWithErr().apply(ParDo.of(new
> DoFn<BigQueryInsertError, String>() {
> // @ProcessElement
> // public void processElement(ProcessContext c) {
> // for(ErrorProto err : c.element().getError().getErrors()){
> // throw new RuntimeException(err.getMessage());
> // }
> //
> // }
> // }));
> result.getFailedInserts().apply(ParDo.of(new DoFn<TableRow, String>() {
> @ProcessElement
> public void processElement(ProcessContext c) {
> System.out.println(c.element());
> c.output("foo");
> throw new RuntimeException("Failed to load");
> }
> @FinishBundle
> public void finishUp(FinishBundleContext finishBundleContextc){
> System.out.println("Got here");
> }
> }));
> pipeline.run();
> } catch (Exception e) {
> e.printStackTrace();
> throw new Exception(e);
> }
> }
> private static TableSchema getTableSchema() {
> List<TableFieldSchema> fields = new ArrayList<>();
> fields.add(new TableFieldSchema().setName("timestamp").setType("INTEGER"));
> fields.add(new TableFieldSchema().setName("payload").setType("STRING"));
> return new TableSchema().setFields(fields);
> }
> public static class RowToBQRow extends DoFn<PubsubMessage, TableRow> {
> @ProcessElement
> public void processElement(ProcessContext c) {
> String payload = new String(c.element().getPayload(),
> StandardCharsets.UTF_8);
> c.output(new TableRow()
> .set("timestamp", System.currentTimeMillis())
> .set("payload", payload)
> );
> }
> }
> }{code}
> Stack trace is attached showing problem of same job id being used/inserted to
> BigQuery on each retry.
> {noformat}
> {noformat}
>
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
