	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Cloud Platforms > Google Cloud Platform (GCP) > Cloud Composer

Cloud Composer

Author: Venkata Sudhakar

Google Cloud Composer is a fully managed workflow orchestration service built on Apache Airflow. It allows you to create, schedule, monitor, and manage data pipelines across cloud and on-premises environments.

Key Features:

1. Apache Airflow - Built on open-source Airflow with full compatibility.

2. Fully managed - Google manages the Airflow infrastructure including upgrades and scaling.

3. GCP integration - Pre-built operators for BigQuery, Dataflow, Dataproc, GCS, and more.

4. Monitoring - Integrated with Cloud Logging and Cloud Monitoring for full observability.

5. Multi-cloud - Orchestrate workflows across GCP, AWS, Azure, and on-premises systems.

The below example shows an Apache Airflow DAG that orchestrates a data pipeline using Cloud Composer.

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.operators.gcs import GCSDeleteObjectsOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-team@company.com"],
}

with DAG(
    dag_id="daily_sales_pipeline",
    default_args=default_args,
    description="Load daily sales data from GCS to BigQuery",
    schedule_interval="0 6 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

# Step 1: Load CSV from GCS to BigQuery
    load_to_bq = GCSToBigQueryOperator(
        task_id="load_sales_to_bigquery",
        bucket="my-data-bucket",
        source_objects=["sales/daily/*.csv"],
        destination_project_dataset_table="my_project.sales.daily_raw",
        schema_fields=[
            {"name": "date", "type": "DATE"},
            {"name": "product_id", "type": "STRING"},
            {"name": "revenue", "type": "FLOAT"},
        ],
        write_disposition="WRITE_TRUNCATE",
    )

# Step 2: Run aggregation query
    aggregate = BigQueryInsertJobOperator(
        task_id="aggregate_daily_sales",
        configuration={
            "query": {
                "query": "INSERT INTO my_project.sales.daily_summary SELECT date, SUM(revenue) as total FROM my_project.sales.daily_raw GROUP BY date",
                "useLegacySql": False,
            }
        },
    )

# Step 3: Clean up raw files
    cleanup = GCSDeleteObjectsOperator(
        task_id="cleanup_raw_files",
        bucket_name="my-data-bucket",
        prefix="sales/daily/",
    )

load_to_bq >> aggregate >> cleanup

It gives the following output in the Airflow UI,

DAG: daily_sales_pipeline
Run: 2024-01-15 06:00:00
Tasks:
  load_sales_to_bigquery   -> SUCCESS (45s)
  aggregate_daily_sales    -> SUCCESS (12s)
  cleanup_raw_files        -> SUCCESS (3s)
Total Duration: 60s

Cloud Composer Environments:

Composer 1 - Based on GKE, uses Airflow 1.x and 2.x. Suitable for existing Airflow workloads.

Composer 2 - Improved performance, auto-scaling, and faster environment creation. Recommended for new deployments.

Send your comments, suggestions or queries regarding this site to [email protected].