Cloud Composer: Job Not Found

[gmyrianthous]

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

BigQueryToGCSOperator(
    task_id='my-task',
    source_project_dataset_table=BQ_TABLE_INPUT,
    destination_cloud_storage_uris=f'{GCS_FOLDER_INPUT}*.jsonl',
    export_format='NEWLINE_DELIMITED_JSON',
    compression=None,
    location='europe-west2'
)

the following error is being raised:

[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

In fact, composer-2.0.13-airflow-2.2.5 was working fine, till I upgraded to composer-2.0.14-airflow-2.2.5

Any clue what may be the issue here and why it does not work on different composer versions?

[potiuk]

I suggest you look at the changelog for the provider: https://airflow.apache.org/docs/apache-airflow-providers-google/stable/index.html#changelog - there are many breaking changes in the way how google providers worked since 1.10 so you will likely find some solution there. Also you might want to look at examples in the docs and maybe yo will find a solution there.

BTW.. Airflow has good and comprehensive documentation that's worth checking on your own before reaching out for help. Strongly recommended.

[gmyrianthous]

@manfredi-giordano Same here 🤔

[gmyrianthous]

Checking to see whether this recent refactoring -or this more recent commit- are related to the issue we are having.

[manfredi-giordano]

@gmyrianthous these look spot on. The change from hook.insert_job to hook.run_extract in the first pull request seems to be what causes the now present warning
WARNING - /opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py:1942: DeprecationWarning: This method is deprecated. Please use `BigQueryHook.insert_job` method. that wasn't being logged before.

The addition of a hook.get_job call after hook.run_extract in the second pull request might be what causes the actual google.api_core.exceptions.NotFound: 404 error. This happens even if, in my experience, the BigQuery job does get created (and succeeds) before the library attempts to retrieve it by ID.

[gmyrianthous]

Yeap, I also noticed that the files are being transferred anyway, but the Task will fail.

[potiuk]

I tihnk this should be fixed in a similar way to: #24165

[manfredi-giordano]

As a workaround, adding a custom Composer dependency pointing to a working version of the google provider works – running the operator doesn't fail anymore.

I've successfully tested it with apache-airflow-providers-google==6.8.0.
Note that because of incompatibilities with packages already installed in Composer clusters, you may need to also define the custom package proto-plus==1.18.1.

[gmyrianthous]

Makes sense.

I wanted to avoid the domino effect of package incompatibilities so for time being I just got rid of the operator and wrote my own PythonOperator to perform the data transfer using the API. 🤔

[shuhoy]

I think the alternative is to use BigQueryInsertJobOperator.

airflow.providers.google.cloud.operators.bigquery.BigQueryInsertJobOperator(
  task_id='your_task',
  configuration = {
        'extract': {
            'sourceTable': {
                'projectId': 'your_project',
                'datasetId': 'your_dataset',
                'tableId': 'your_table',
            },
            'compression': None,
            'destinationUris': 'gs://you_bucket/.../file_name.jsonl',
            'destinationFormat': 'NEWLINE_DELIMITED_JSON'
        }
    },
  gcp_conn_id='your_conn_id',
)

[potiuk]

FYI. I just merged a workaround for now (not a proper fix yet - as the deprecated function remains) in #24330 and it will be available in RC2 of the new google provider I am releasing (8.0.0).

[karakanb]

Just upgraded to Airflow v2.4.1 with v8.4.0, the issue persists, BigQueryValueCheckOperator doesn't work without the location specified, even though all the other BigQuery operators work properly. Is there another forum that I can report this to help getting it resolved / collaborate to resolve it myself?

[potiuk]

You can create PR.

[potiuk]

Or open issue. We will mark it as 'good-first-issue" and maybe someone will pick it up.

[potiuk]

Just make sure to describe all details.

[gmyrianthous]

@karakanb In the meantime, you can follow this workaround: https://stackoverflow.com/a/72548247/7131757