LinkedIn
Twitter
Dear Google Cloud Community,
I hope this message finds you well.
I am encountering an issue while using the DataformCreateWorkflowInvocationOperator in Apache Airflow, specifically when attempting to trigger a workflow invocation in Dataform. The error message I am seeing is:
google.api_core.exceptions.InvalidArgument: 400 At least one action must be selected for execution.
This issue arises even though I have already included valid actions (such as "compile" and "run") in the "invocation_config" section of the operator. Here's the relevant part of my Airflow DAG code:
Despite ensuring that the "actions" field is properly populated with the actions "compile" and "run", I continue to receive the same error message, indicating that no actions are selected for execution.
Could you kindly assist me in resolving this issue? Is there a specific format or configuration for the actions field that I might be missing?
ERROR LOGS
default-hostname
*** Reading remote logs from Cloud Logging.
[2025-04-02 07:46:30.565966+00:00] {local_task_job_runner.py:123} INFO - ::group::Pre task execution logs
[2025-04-02 07:46:30.623490+00:00] {taskinstance.py:2615} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: dataform.create_workflow_invocation manual__2025-04-02T07:46:24.346933+00:00 [queued]>
[2025-04-02 07:46:30.643707+00:00] {taskinstance.py:2615} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: dataform.create_workflow_invocation manual__2025-04-02T07:46:24.346933+00:00 [queued]>
[2025-04-02 07:46:30.644272+00:00] {taskinstance.py:2868} INFO - Starting attempt 1 of 3
[2025-04-02 07:46:30.673532+00:00] {taskinstance.py:2891} INFO - Executing <Task(DataformCreateWorkflowInvocationOperator): create_workflow_invocation> on 2025-04-02 07:46:24.346933+00:00
[2025-04-02 07:46:30.692012+00:00] {standard_task_runner.py:72} INFO - Started process 7200 to run task
[2025-04-02 07:46:30.697426+00:00] {standard_task_runner.py:104} INFO - Running: ['airflow', 'tasks', 'run', 'dataform', 'create_workflow_invocation', 'manual__2025-04-02T07:46:24.346933+00:00', '--job-id', '54', '--raw', '--subdir', 'DAGS_FOLDER/dataforms_123.py', '--cfg-path', '/tmp/tmp3j538_qp']
[2025-04-02 07:46:30.698234+00:00] {standard_task_runner.py:105} INFO - Job 54: Subtask create_workflow_invocation
[2025-04-02 07:46:30.803587+00:00] {task_command.py:473} INFO - Running <TaskInstance: dataform.create_workflow_invocation manual__2025-04-02T07:46:24.346933+00:00 [running]> on host airflow-worker-lxt2n
[2025-04-02 07:46:31.555827+00:00] {taskinstance.py:3134} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='dataform' AIRFLOW_CTX_TASK_ID='create_workflow_invocation' AIRFLOW_CTX_EXECUTION_DATE='2025-04-02T07:46:24.346933+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2025-04-02T07:46:24.346933+00:00'
[2025-04-02 07:46:31.556913+00:00] {taskinstance.py:732} INFO - ::endgroup::
[2025-04-02 07:46:31.634186+00:00] {base.py:84} INFO - Retrieving connection 'google_cloud_default'
[2025-04-02 07:46:31.635419+00:00] {credentials_provider.py:409} INFO - Getting connection using `google.auth.default()` since no explicit credentials are provided.
[2025-04-02 07:46:31.682999+00:00] {taskinstance.py:3313} ERROR - Task failed with exception\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 36, in to_proto\n return self._descriptor(**value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 36, in to_proto\n return self._descriptor(**value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 768, in _execute_task\n result = _execute_callable(context=context, **execute_callable_kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 734, in _execute_callable\n return ExecutionCallableRunner(\n ^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/utils/operator_helpers.py", line 252, in run\n return self.func(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 415, in wrapper\n return func(self, *args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/cloud/operators/dataform.py", line 248, in execute\n result = hook.create_workflow_invocation(\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/common/hooks/base_google.py", line 554, in inner_wrapper\n return func(self, *args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/cloud/hooks/dataform.py", line 189, in create_workflow_invocation\n return client.create_workflow_invocation(\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/google/cloud/dataform_v1beta1/services/dataform/client.py", line 5897, in create_workflow_invocation\n request = dataform.CreateWorkflowInvocationRequest(request)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 728, in __init__\n pb_value = marshal.to_proto(pb_type, value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/marshal.py", line 235, in to_proto\n pb_value = self.get_rule(proto_type=proto_type).to_proto(value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 46, in to_proto\n return self._wrapper(value)._pb\n ^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 728, in __init__\n pb_value = marshal.to_proto(pb_type, value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/marshal.py", line 235, in to_proto\n pb_value = self.get_rule(proto_type=proto_type).to_proto(value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 46, in to_proto\n return self._wrapper(value)._pb\n ^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 734, in __init__\n super().__setattr__("_pb", self._meta.pb(**params))\n ^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.
[2025-04-02 07:46:31.729521+00:00] {taskinstance.py:1226} INFO - Marking task as UP_FOR_RETRY. dag_id=dataform, task_id=create_workflow_invocation, run_id=manual__2025-04-02T07:46:24.346933+00:00, execution_date=20250402T074624, start_date=20250402T074630, end_date=20250402T074631
[2025-04-02 07:46:31.761009+00:00] {taskinstance.py:340} INFO - ::group::Post task execution logs
[2025-04-02 07:46:31.767888+00:00] {standard_task_runner.py:124} ERROR - Failed to execute job 54 for task create_workflow_invocation (Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.; 7200)\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 36, in to_proto\n return self._descriptor(**value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 36, in to_proto\n return self._descriptor(**value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/task/task_runner/standard_task_runner.py", line 117, in _start_by_fork\n ret = args.func(args, dag=self.dag)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/cli/cli_config.py", line 49, in command\n return func(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/utils/cli.py", line 115, in wrapper\n return f(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/cli/commands/task_command.py", line 489, in task_run\n task_return_code = _run_task_by_selected_method(args, _dag, ti)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/cli/commands/task_command.py", line 256, in _run_task_by_selected_method\n return _run_raw_task(args, ti)\n ^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/cli/commands/task_command.py", line 341, in _run_raw_task\n return ti._run_raw_task(\n ^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/utils/session.py", line 97, in wrapper\n return func(*args, session=session, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 3007, in _run_raw_task\n return _run_raw_task(\n ^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 273, in _run_raw_task\n TaskInstance._execute_task_with_callbacks(\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 3161, in _execute_task_with_callbacks\n result = self._execute_task(context, task_orig)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 3185, in _execute_task\n return _execute_task(self, context, task_orig)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 768, in _execute_task\n result = _execute_callable(context=context, **execute_callable_kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 734, in _execute_callable\n return ExecutionCallableRunner(\n ^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/utils/operator_helpers.py", line 252, in run\n return self.func(*args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 415, in wrapper\n return func(self, *args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/cloud/operators/dataform.py", line 248, in execute\n result = hook.create_workflow_invocation(\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/common/hooks/base_google.py", line 554, in inner_wrapper\n return func(self, *args, **kwargs)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/airflow/providers/google/cloud/hooks/dataform.py", line 189, in create_workflow_invocation\n return client.create_workflow_invocation(\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/google/cloud/dataform_v1beta1/services/dataform/client.py", line 5897, in create_workflow_invocation\n request = dataform.CreateWorkflowInvocationRequest(request)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 728, in __init__\n pb_value = marshal.to_proto(pb_type, value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/marshal.py", line 235, in to_proto\n pb_value = self.get_rule(proto_type=proto_type).to_proto(value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 46, in to_proto\n return self._wrapper(value)._pb\n ^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 728, in __init__\n pb_value = marshal.to_proto(pb_type, value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/marshal.py", line 235, in to_proto\n pb_value = self.get_rule(proto_type=proto_type).to_proto(value)\n ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/marshal/rules/message.py", line 46, in to_proto\n return self._wrapper(value)._pb\n ^^^^^^^^^^^^^^^^^^^^\n File "/opt/python3.11/lib/python3.11/site-packages/proto/message.py", line 734, in __init__\n super().__setattr__("_pb", self._meta.pb(**params))\n ^^^^^^^^^^^^^^^^^^^^^^^\nTypeError: Parameter to MergeFrom() must be instance of same class: expected <class 'Target'> got <class 'str'>.
[2025-04-02 07:46:31.800010+00:00] {local_task_job_runner.py:266} INFO - Task exited with return code 1
[2025-04-02 07:46:31.860284+00:00] {taskinstance.py:3904} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2025-04-02 07:46:31.863627+00:00] {local_task_job_runner.py:245} INFO - ::endgroup::
Any help or insights would be greatly appreciated!
Thank you for your time and support.
Attached Snapshot for the reference.
