BigQueryExecuteQueryOperator timeout issue (how to increase timeout)

user_7726196 0
Viewed 578

faced with an issue in Airflow. In short: I trying to move from 1.10.14 version to 2.2.2 ver. However when I want to substitute BigQueryOperator to BigQueryExecuteQueryOperator I tried to increase jobTimeoutMs how Bigquery API docs are told but still see this issue.

I observe this issue when the job runs longer than 1 minute Meanwhile, in the old Bigquery operator with the identically configured connection for google cloud on Airflow 1.10.14 I haven't seen this kind of issue

    mobile_push_stat = BigQueryExecuteQueryOperator(
        task_id="mobile_push_stat",
        sql="/sql/updater/mobile_push_stat.sql",
        use_legacy_sql=False,
        api_resource_configs={"jobTimeoutMs": "3600000"},
        gcp_conn_id="bigquery_work",
    )

LOGs

*** Reading local file: /srv/airflow/logs/ExtensionPushStat/mobile_push_stat/2021-12-05T06:00:00+00:00/29.log
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1035} INFO - Dependencies all met for <TaskInstance: ExtensionPushStat.mobile_push_stat scheduled__2021-12-05T06:00:00+00:00 [queued]>
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1035} INFO - Dependencies all met for <TaskInstance: ExtensionPushStat.mobile_push_stat scheduled__2021-12-05T06:00:00+00:00 [queued]>
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1241} INFO - 
--------------------------------------------------------------------------------
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1242} INFO - Starting attempt 29 of 29
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1243} INFO - 
--------------------------------------------------------------------------------
[2021-12-06, 17:33:11 UTC] {taskinstance.py:1262} INFO - Executing <Task(BigQueryExecuteQueryOperator): mobile_push_stat> on 2021-12-05 06:00:00+00:00
[2021-12-06, 17:33:11 UTC] {base_task_runner.py:141} INFO - Running on host: airflow-vm-v2
[2021-12-06, 17:33:11 UTC] {base_task_runner.py:142} INFO - Running: ['airflow', 'tasks', 'run', 'ExtensionPushStat', 'mobile_push_stat', 'scheduled__2021-12-05T06:00:00+00:00', '--job-id', '3235', '--raw', '--subdir', 'DAGS_FOLDER/ExtensionPushStat.py', '--cfg-path', '/tmp/tmpk6p4qql0', '--error-file', '/tmp/tmpwz3z5o0y']
[2021-12-06, 17:33:12 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat [[34m2021-12-06, 17:33:12 UTC[0m] {[34mdagbag.py:[0m500} INFO[0m - Filling up the DagBag from /srv/airflow/dags/ExtensionPushStat.py[0m
[2021-12-06, 17:34:11 UTC] {local_task_job.py:206} WARNING - Recorded pid 137796 does not match the current pid 137817
[2021-12-06, 17:34:11 UTC] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 137817
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat Running <TaskInstance: ExtensionPushStat.mobile_push_stat scheduled__2021-12-05T06:00:00+00:00 [running]> on host airflow-vm-v2
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat Traceback (most recent call last):
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/bin/airflow", line 8, in <module>
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     sys.exit(main())
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/__main__.py", line 48, in main
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     args.func(args)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 48, in command
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return func(*args, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/utils/cli.py", line 92, in wrapper
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return f(*args, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 292, in task_run
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     _run_task_by_selected_method(args, dag, ti)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     _run_raw_task(args, ti)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 180, in _run_raw_task
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     ti._run_raw_task(
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/utils/session.py", line 70, in wrapper
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return func(*args, session=session, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     self._execute_task_with_callbacks(context)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     result = self._execute_task(context, self.task)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     result = execute_callable(context=context)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/providers/google/cloud/operators/bigquery.py", line 693, in execute
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     job_id = self.hook.run_query(
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 2325, in run_query
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     job = self.insert_job(configuration=configuration, project_id=self.project_id)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/providers/google/common/hooks/base_google.py", line 425, in inner_wrapper
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return func(self, *args, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/bigquery.py", line 1639, in insert_job
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     job.result()
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/cloud/bigquery/job/query.py", line 1450, in result
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     do_get_result()
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/cloud/bigquery/job/query.py", line 1440, in do_get_result
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     super(QueryJob, self).result(retry=retry, timeout=timeout)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/cloud/bigquery/job/base.py", line 727, in result
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return super(_AsyncJob, self).result(timeout=timeout, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/api_core/future/polling.py", line 130, in result
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     self._blocking_poll(timeout=timeout, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/cloud/bigquery/job/query.py", line 1199, in _blocking_poll
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     super(QueryJob, self)._blocking_poll(timeout=timeout, **kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/api_core/future/polling.py", line 108, in _blocking_poll
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     retry_(self._done_or_raise)(**kwargs)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/api_core/retry.py", line 286, in retry_wrapped_func
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     return retry_target(
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/google/api_core/retry.py", line 220, in retry_target
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     time.sleep(sleep)
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat   File "/srv/airflow/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1413, in signal_handler
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat     raise AirflowException("Task received SIGTERM signal")
[2021-12-06, 17:34:11 UTC] {base_task_runner.py:122} INFO - Job 3235: Subtask mobile_push_stat airflow.exceptions.AirflowException: Task received SIGTERM signal
[2021-12-06, 17:34:11 UTC] {process_utils.py:66} INFO - Process psutil.Process(pid=137817, status='terminated', exitcode=1, started='17:33:11') (137817) terminated with exit code 1

UPDATE

This is issue that does not relate to Bigquery or Google providers.

This issue related to two config parameters

killed_task_cleanup_time - in this case for long-running jobs needs to be increased job_heartbeat_sec - in this case when you want to clear status for the failed task and re-run from UI

airflow google-bigquery
1 Answers

BigQueryExecuteQueryOperator is deprecated (see source code). You should use BigQueryInsertJobOperator.

Based on api parameters and the operator source code the syntax should be:

mobile_push_stat = BigQueryInsertJobOperator(
    task_id="mobile_push_stat",
    gcp_conn_id="bigquery_work",
    configuration={
        "query": "/sql/updater/mobile_push_stat.sql",
        "useLegacySql": False,
        "timeoutMs": 3600000,

    },
)

Edit: Note the limitations around the timeoutMs. The value you choose 3600000 is probably too high.

You can request a longer timeout period in the timeoutMs field. However, the call is not guaranteed to wait for the specified timeout; it typically returns after around 200 seconds (200,000 milliseconds), even if the query is not complete.

user_146244090
Related