title: How to transform data before loading into BigQuery in Apache Airflow?
body: <p>I am new to Apache Airflow. My task is to read data from Google Cloud Storage, transform the data and upload the transformed data into BigQuery table. I'm able to get data from Cloud Storage bucket and directly store that to BigQuery table. I'm not sure how to include the transform function in this pipeline.</p>
  <p>Here's my code:</p>
  <pre><code># Import libraries needed for the operation
  import airflow
  from datetime import timedelta, datetime
  from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.contrib.operators.bigquery_operator import BigQueryOperator
  from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
  
  # Default Argument
  default_args = {
      'owner': &lt;OWNER_NAME&gt;,
      'depends_on_past': False,
      'start_date': airflow.utils.dates.days_ago(1),
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 2,
      'retry_delay': timedelta(minutes=2),
  }
  
  # DAG Definition
  dag = DAG('load_from_bucket_to_bq',
  schedule_interval='0 * * * *',
  default_args=default_args)
  
  # Variable Configurations
  BQ_CONN_ID = &lt;CONN_ID&gt;
  BQ_PROJECT = &lt;PROJECT_ID&gt;
  BQ_DATASET = &lt;DATASET_ID&gt;
  
  with dag:
      # Tasks
      start = DummyOperator(
          task_id='start'
      )
  
      upload = GoogleCloudStorageToBigQueryOperator(
          task_id='load_from_bucket_to_bigquery',
          bucket=&lt;BUCKET_NAME&gt;,
          source_objects=['*.csv'],
          schema_fields=[
              {'name': 'Active_Cases', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'Country', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'Last_Update', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'New_Cases', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'New_Deaths', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'Total_Cases', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'Total_Deaths', 'type': 'STRING', 'mode': 'NULLABLE'},
              {'name': 'Total_Recovered', 'type': 'STRING', 'mode': 'NULLABLE'},
          ],
          destination_project_dataset_table=BQ_PROJECT + '.' + BQ_DATASET + '.' + &lt;TABLE_NAME&gt;,
          write_disposition='WRITE_TRUNCATE',
          google_cloud_storage_conn_id=BQ_CONN_ID,
          bigquery_conn_id=BQ_CONN_ID,
          dag = dag
      )
  
      end = DummyOperator(
          task_id='end'
      )
  
      # Setting Dependencies
      start &gt;&gt; upload &gt;&gt; end
  
  </code></pre>
  <p>Any help on how to proceed is appreciated. Thanks.</p>