Airflow 2: Job Not Found when transferring data from BigQuery into Cloud Storage

user_7131757 0
Viewed 430

I am trying to migrate from Cloud Composer 1 into Cloud Composer 2 (from Airflow 1.10.15 into Airflow 2.2.5) and when attempting to load data from BigQuery into GCS using the BigQueryToGCSOperator

from airflow.providers.google.cloud.transfers.bigquery_to_gcs import BigQueryToGCSOperator 

# ...

BigQueryToGCSOperator(
    task_id='my-task',
    source_project_dataset_table='my-project-name.dataset-name.table-name',
    destination_cloud_storage_uris=f'gs://my-bucket/another-path/*.jsonl',
    export_format='NEWLINE_DELIMITED_JSON',
    compression=None,
    location='europe-west2'
)

that results into the following error:

[2022-06-07, 11:17:01 UTC] {taskinstance.py:1776} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py", line 141, in execute
    job = hook.get_job(job_id=job_id).to_api_repr()
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/common/hooks/base_google.py", line 439, in inner_wrapper
    return func(self, *args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 1492, in get_job
    job = client.get_job(job_id=job_id, project=project_id, location=location)
  File "/opt/python3.8/lib/python3.8/site-packages/google/cloud/bigquery/client.py", line 2066, in get_job
    resource = self._call_api(
  File "/opt/python3.8/lib/python3.8/site-packages/google/cloud/bigquery/client.py", line 782, in _call_api
    return call()
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/retry.py", line 283, in retry_wrapped_func
    return retry_target(
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/retry.py", line 190, in retry_target
    return target()
  File "/opt/python3.8/lib/python3.8/site-packages/google/cloud/_http/__init__.py", line 494, in api_request
    raise exceptions.from_http_response(response)

google.api_core.exceptions.NotFound: 404 GET https://bigquery.googleapis.com/bigquery/v2/projects/my-project-name/jobs/airflow_1654592634552749_1896245556bd824c71f31c79d28cdfbe?projection=full&prettyPrint=false: Not found: Job my-project-name:airflow_1654592634552749_1896245556bd824c71f31c79d28cdfbe

Any clue what may be the issue here and why it does not work on Airflow 2.2.5 (even though the equivalent BigQueryToCloudStorageOperator works for Cloud Composer v1 in Airflow 1.10.15).

google-cloud-composer airflow google-cloud-storage google-cloud-platform
1 Answers

Apparently this seems to be a bug introduced in apache-airflow-providers-google version v7.0.0.

Also note that the file transfer from BQ into GCS will actually be successful (even though the task will fail).

As a workaround you can either revert back to a working version (if this is possible) e.g. to 6.8.0, or make use of the BQ API and get rid of BigQueryToGCSOperator.

For example,

from google.cloud import bigquery
from airflow.operators.python import PythonOperator


def load_bq_to_gcs():
    client = bigquery.Client()
    job_config = bigquery.job.ExtractJobConfig()
    job_config.destination_format = bigquery.DestinationFormat.NEWLINE_DELIMITED_JSON

    destination_uri = f"{<gcs-bucket-destination>}*.jsonl"

    dataset_ref = bigquery.DatasetReference(bq_project_name, bq_dataset_name)
    table_ref = dataset_ref.table(bq_table_name)

    extract_job = client.extract_table(
        table_ref,
        destination_uri,
        job_config=job_config,
        location='europe-west2',
    )
    extract_job.result()

and then create an instance of PythonOperator:

PythonOperator(
    task_id='test_task',
    python_callable=load_bq_to_gcs,
)
user_71317570
Related