[I] Using XComArg in BigQueryInsertJobOperator `params` [airflow]

["Locustv2 (via GitHub)" <gi...@apache.org>]

Locustv2 opened a new issue, #39914:
URL: https://github.com/apache/airflow/issues/39914

   ### Description
   
   Hey guys, 
   
   So i have been dealing with this issue for a while now without any light...
   I have a DAG that queries data from BigQuery, and depending on the results some Dynamic Task Mappings are created to insert an entry in another BigQuery table using `BigQueryInsertJobOperator`...
   
   For Example:
   
   ```
   from airflow import DAG
   from airflow.providers.google.cloud.operators.bigquery import BigQueryGetDataOperator, BigQueryInsertJobOperator
   from airflow.utils.dates import days_ago
   from airflow.decorators import task
   from airflow import XComArg
   
   default_args = {
       'owner': 'airflow',
       'start_date': days_ago(1),
       'retries': 1,
   }
   
   dag = DAG(
       dag_id='bigquery_data_transfer_mapped_correct',
       default_args=default_args,
       schedule_interval="@daily",
       catchup=False,
       tags=['example'],
   )
   
     @task
     def get_data(sql):
         bq_hook = BigQueryHook(...)
     
         self.log.info('Fetching Data from:')
         self.log.info('Query: %s', sql)
     
         bq_client = bq_hook.get_client()
         query_job = bq_client.query(sql)
         client_results = query_job.result()  # Waits for the query to finish
     
         results = list(dict(result) for  result in client_results)
     
         self.log.info(f"Retrieved {len(results)} rows from BigQuery")
         self.log.info('Response: %s', results)
     
         return results
   
     query_data = get_data("SELECT * FROM some_table WHERE some_conditions;")
     
     insert = BigQueryInsertJobOperator.partial(
         task_id=f"insert_data",
         configuration={
             'query': {
                 'query': "INSERT INTO `project.dataset.table` (field1, field2) VALUES ('{{ params.field1 }}', '{{ params.field2 }}')",
                 'useLegacySql': False,
             }
         }
     ).expand(
         params=XComArg(query_data)
     )
     
     query_data >> insert
   
   ```
   
   Please note that this code is just a basic example that i just wrote and in my usecase, i actually have a task_group that expands and takes in a parameter to be sent to the `params` in one of the `BigQueryInsertJobOperator` task.
   
   
   After running my DAG i get an error saying:
   
   ```
   Broken DAG: [/opt/airflow/dags/src/dag.py] Traceback (most recent call last):
     File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 407, in apply_defaults
       default_args, merged_params = get_merged_defaults(
                                     ^^^^^^^^^^^^^^^^^^^^
     File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 167, in get_merged_defaults
       raise TypeError("params must be a mapping")
   TypeError: params must be a mapping
   ```
   
   The airflow version is: 
   
   ```
   Version: [v2.8.1](https://pypi.python.org/pypi/apache-airflow/2.8.1)
   Git Version: .release:c0ffa9c5d96625c68ded9562632674ed366b5eb3
   ```
   
   ### Use case/motivation
   
   I would like to be able to use dynamic task mapping to create multiple instances of `BigQueryInsertJobOperator` with dynamic `params` attribute.
   
   ### Related issues
   
   _No response_
   
   ### Are you willing to submit a PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Using XComArg in BigQueryInsertJobOperator `params` [airflow]

["boring-cyborg[bot] (via GitHub)" <gi...@apache.org>]

boring-cyborg[bot] commented on issue #39914:
URL: https://github.com/apache/airflow/issues/39914#issuecomment-2137465998

   Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Using XComArg in BigQueryInsertJobOperator `params` [airflow]

["josh-fell (via GitHub)" <gi...@apache.org>]

josh-fell closed issue #39914: Using XComArg in BigQueryInsertJobOperator `params`
URL: https://github.com/apache/airflow/issues/39914


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org