Variable passing throwing error in BigQueryInsertJobOperator in Airflow

user_1144868 0
Viewed 158

I have written a BigQueryInsertJobOperator in Airflow to select and insert data to a Big Query table. But I am facing issue with variable passing. I am getting below error while executing Airflow DAG.

File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job/query.py", line 911, in to_api_repr configuration = self._configuration.to_api_repr() File "/home/airflow/.local/lib/python3.7/site-packages/google/cloud/bigquery/job/query.py", line 683, in to_api_repr query_parameters = resource["query"].get("queryParameters") AttributeError: 'str' object has no attribute 'get'

Here is my Operator code:

dag = DAG(
'bq_to_sql_operator', 
default_args=default_args, 
schedule_interval="@daily",
template_searchpath="/opt/airflow/dags/scripts",
user_defined_macros={"BQ_PROJECT": BQ_PROJECT, "BQ_EDW_DATASET": BQ_EDW_DATASET, "BQ_STAGING_DATASET": BQ_STAGING_DATASET},
catchup=False
)

t1 = BigQueryInsertJobOperator(
    task_id='bq_write_to_umc_cg_service_agg_stg',
    configuration={
        "query": "{% include 'umc_cg_service_agg_stg.sql' %}",
        "useLegacySql":False,
        "allow_large_results":True,
        "writeDisposition": "WRITE_TRUNCATE", 
        "destinationTable": {
            'projectId': BQ_PROJECT,
            'datasetId': BQ_STAGING_DATASET,
            'tableId': UMC_CG_SERVICE_AGG_STG_TABLE_NAME
        }
    },
    params={'BQ_PROJECT': BQ_PROJECT, 'BQ_EDW_DATASET': BQ_EDW_DATASET, 'BQ_STAGING_DATASET': BQ_STAGING_DATASET },
    gcp_conn_id=BQ_CONN_ID,
    location=BQ_LOCATION,
    dag=dag
)

My SQL file looks like as below:

select
        faccs2.employer_key employer_key,
        faccs2.service_name service_name,
        gender,
        approximate_age_band,
        state,
        relationship_map_name,
        account_attribute1_name,
        account_attribute1_value,
        account_attribute2_name,
        account_attribute2_value,
        account_attribute3_name,
        account_attribute3_value,
        account_attribute4_name,
        account_attribute4_value,
        account_attribute5_name,
        account_attribute5_value,
        count(distinct faccs2.sf_service_id) total_service_count
    from `{{params.BQ_PROJECT}}.{{params.BQ_EDW_DATASET}}.fact_account_cg_case_survey` faccs
    inner join `{{params.BQ_PROJECT}}.{{params.BQ_EDW_DATASET}}.fact_account_cg_case_service` faccs2 on faccs.sf_case_id = faccs2.sf_case_id
    inner join `{{params.BQ_PROJECT}}.{{params.BQ_EDW_DATASET}}.dim_account` da on faccs2.account_key = da.account_key
    left join `{{params.BQ_PROJECT}}.{{params.BQ_STAGING_DATASET}}.stg_account_selected_attr_tmp2` attr on faccs.account_key = attr.account_key
    where not da.is_test_account_flag
    and attr.gender is not null
    and coalesce(faccs.case_status,'abc') <> 'Closed as Duplicate'
    group by 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16;

Can someone please help me how to fix this issue.

airflow google-bigquery directed-acyclic-graphs google-cloud-platform
2 Answers

I think that the query configuration should be in a nested document called query:

t1 = BigQueryInsertJobOperator(
    task_id='bq_write_to_umc_cg_service_agg_stg',
    configuration={
        "query": {
            "query": "{% include 'umc_cg_service_agg_stg.sql' %}",
            "useLegacySql":False,
            "allow_large_results":True,
            "writeDisposition": "WRITE_TRUNCATE", 
            "destinationTable": {
                'projectId': BQ_PROJECT,
                'datasetId': BQ_STAGING_DATASET,
                'tableId': UMC_CG_SERVICE_AGG_STG_TABLE_NAME
            }
        }
    },
    params={'BQ_PROJECT': BQ_PROJECT, 'BQ_EDW_DATASET': BQ_EDW_DATASET, 'BQ_STAGING_DATASET': BQ_STAGING_DATASET },
    gcp_conn_id=BQ_CONN_ID,
    location=BQ_LOCATION,
    dag=dag
)

With your provided configuration dict, an internal method try to access queryParameters which should be in the dict configuration["query"], but it finds str instead of dict.

user_95605940

Consider below script what I've used at work.

    target_date = '{{ ds_nodash }}'

    ...

    # DAG task
    t1= bq.BigQueryInsertJobOperator(
        task_id = 'sample_task,
        configuration = {
            "query": {
                "query": f"{{% include 'your_query_file.sql' %}}",
                "useLegacySql": False,
                "queryParameters": [
                    { "name": "target_date", 
                      "parameterType": { "type": "STRING" },
                      "parameterValue": { "value": f"{target_date}" }
                    }
                ],
                "parameterMode": "NAMED"
            },
        },
        location = 'asia-northeast3',
    )

-- in your_query_file.sql, @target_date value is passed as a named parameter.

DECLARE target_date DATE DEFAULT SAFE.PARSE_DATE('%Y%m%d', @target_date);

SELECT ... FROM ... WHERE partitioned_at = target_date;

You can refer to configuration JSON field specification on the link below.

parameterMode string

Standard SQL only. Set to POSITIONAL to use positional (?) query parameters or to NAMED to use named (@myparam) query parameters in this query.

queryParameters[] object (QueryParameter)

jobs.query parameters for Standard SQL queries.

queryParameters is an array of QueryParameter which has following JSON format.

{
  "name": string,
  "parameterType": {
    object (QueryParameterType)
  },
  "parameterValue": {
    object (QueryParameterValue)
  }
}
user_190399200
Related