Google dataflow API call failure-

Question

I am trying to call the dataflow job from composer airflow using the dataflow operator but getting the below error while calling it:

googleapiclient.errors.HttpError: <HttpError 400 when requesting https://test returned "Invalid value at 'launch_parameters.parameters' (type.googleapis.com/google.dataflow.v1beta3.LaunchTemplateParameters.ParametersEntry), "{'test1': 'SELECT distinct data\nFROM project.dataset.table1\nWHERE ace_date="2022-05-12"', 'test2': 'SELECT distinct data\nFROM project.dataset.table2\nWHERE ace_date="2022-05-12"', 'priority_data': 'SELECT distinct data\nFROM project.dataset.table3\nWHERE ace_date="2022-05-12"', 'test3': 'SELECT distinct data\nFROM project.dataset.table4\nWHERE ace_date="2022-05-12"', 'test4': 'SELECT distinct data\nFROM project.dataset.table5\nWHERE ace_date="2022-05-12"', 'test5': 'SELECT distinct data\nFROM project.dataset.tabl6\nWHERE ace_date="2022-05-12"', 'pack_rules': 'SELECT distinct data\nFROM project.dataset.table7\nWHERE ace_date="2022-05-12"', 'test6': 'SELECT distinct row_key_data as data\nFROM peoject.dataset.table7\nWHERE date_of_run="2022-05-16"'}""

Below is code for the same while calling it from Airflow:

def dataflow_trigger(
        task,
):
    """
    Dynamic task for calling dataflow job
    """
    return DataflowTemplatedJobStartOperator(
        task_id=task,
        project_id="{{task_instance.xcom_pull(key='dataflow_settings', task_ids='get_settings')['project']}}",
        job_name="{{task_instance.xcom_pull(key='dataflow_settings', task_ids='get_settings')['job_name']}}",
        template="{{task_instance.xcom_pull(key='dataflow_settings', task_ids='get_settings')['template_path']}}",
        parameters="{{task_instance.xcom_pull(key='parameters', task_ids='get_settings')}}",
        location='europe-west2',
    )

score 0 · Accepted Answer · answered May 16 '22 at 17:59

0

Airflow xcom pull only returns string This helped to fix the issue as xcom push stores it as a string; used render_template_as_native_obj=True in DAG constructor solved this.

answered May 16 '22 at 17:59

Krishnamoorthy Ganapathy

21
6

As it’s currently written, your answer is unclear. Please [edit] to add additional details that will help others understand how this addresses the question asked. You can find more information on how to write good answers [in the help center](/help/how-to-answer). – Community May 18 '22 at 12:32

Google dataflow API call failure-

1 Answers1