google-cloud-data-engineering-hub

# Google Cloud Data Engineering Hub Skill > Skill by [ara.so](https://ara.so) — Data Skills collection This skill enables AI coding agents to help developers build production-grade Google Cloud Platform (GCP) data engineering solutions using this comprehensive reference repository of 54+ working projects covering BigQuery, Dataflow, Apache Beam, Cloud Composer, Pub/Sub, Dataproc, Gemini AI, and Vertex AI. ## What This Project Provides A curated collection of complete, runnable GCP data engineering projects. Each project includes: - Modular Python code (not scripts) - ASCII architecture diagrams - Sample data fixtures - `deploy.sh` with GCP setup automation - End-to-end working examples tested against live GCP environments Built by Vishal Bulbule (Google Developer Expert, 12x GCP Certified). ## Installation & Setup ### Clone the Repository ```bash git clone https://github.com/vishal-bulbule/google-cloud-data-engineering-hub.git cd google-cloud-data-engineering-hub ``` ### Prerequisites - Python 3.10+ - Google Cloud SDK (`gcloud` CLI) - Active GCP project with billing enabled - Appropriate IAM permissions ### Environment Setup ```bash # Set up Python virtual environment python3 -m venv .venv source .venv/bin/activate # On Windows: .venv\Scripts\activate # Install dependencies (per project) cd <project-directory> pip install -r requirements.txt # Configure GCP credentials export GOOGLE_CLOUD_PROJECT=your-project-id export GOOGLE_CLOUD_LOCATION=us-central1 gcloud auth application-default login ``` ## Project Categories & Key Examples ### BigQuery Projects (01-07) #### CSV Ingestion Pipeline (01-bq-csv-ingestion-pipeline) ```python from google.cloud import bigquery def load_csv_to_bigquery( project_id: str, dataset_id: str, table_id: str, csv_file_path: str ): """Load CSV from local disk to BigQuery.""" client = bigquery.Client(project=project_id) table_ref = f"{project_id}.{dataset_id}.{table_id}" job_config = bigquery.LoadJobConfig( source_format=bigquery.SourceFormat.CSV, skip_leading_rows=1, autodetect=True, write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE, ) with open(csv_file_path, "rb") as source_file: job = client.load_table_from_file( source_file, table_ref, job_config=job_config ) job.result() # Wait for completion print(f"Loaded {job.output_rows} rows into {table_ref}") ``` #### UPSERT/MERGE Pattern (03-bq-upsert-merge-pattern) ```python from google.cloud import bigquery def upsert_data(project_id: str, dataset_id: str, table_id: str): """Perform MERGE operation for upsert pattern.""" client = bigquery.Client(project=project_id) merge_query = f""" MERGE `{project_id}.{dataset_id}.{table_id}` AS target USING `{project_id}.{dataset_id}.staging_table` AS source ON target.id = source.id WHEN MATCHED THEN UPDATE SET name = source.name, value = source.value, updated_at = CURRENT_TIMESTAMP() WHEN NOT MATCHED THEN INSERT (id, name, value, created_at, updated_at) VALUES (source.id, source.name, source.value, CURRENT_TIMESTAMP(), CURRENT_TIMESTAMP()) """ query_job = client.query(merge_query) result = query_job.result() print(f"MERGE completed. Rows modified: {result.total_rows}") ``` #### BigQuery ML (05-bq-ml-train-predict) ```python from google.cloud import bigquery def train_ml_model(project_id: str, dataset_id: str): """Train a logistic regression model using BigQuery ML.""" client = bigquery.Client(project=project_id) training_query = f""" CREATE OR REPLACE MODEL `{project_id}.{dataset_id}.classification_model` OPTIONS( model_type='LOGISTIC_REG', input_label_cols=['label'], max_iterations=10 ) AS SELECT feature1, feature2, feature3, label FROM `{project_id}.{dataset_id}.training_data` """ job = client.query(training_query) job.result() print("Model training completed") def predict_with_model(project_id: str, dataset_id: str): """Make predictions using trained BQML model.""" client = bigquery.Client(project=project_id) prediction_query = f""" SELECT * FROM ML.PREDICT( MODEL `{project_id}.{dataset_id}.classification_model`, (SELECT feature1, feature2, feature3 FROM `{project_id}.{dataset_id}.prediction_data`) ) """ results = client.query(prediction_query).to_dataframe() return results ``` ### Cloud Storage Projects (08-10) #### File Management (08-gcs-file-management) ```python from google.cloud import storage def upload_blob(bucket_name: str, source_file: str, destination_blob: str): """Upload a file to GCS.""" storage_client = storage.Client() bucket = storage_client.bucket(bucket_name) blob = bucket.blob(destination_blob) blob.upload_from_filename(source_file) print(f"File {source_file} uploaded to {destination_blob}") def list_blobs_with_prefix(bucket_name: str, prefix: str): """List all blobs with a specific prefix.""" storage_client = storage.Client() blobs = storage_client.list_blobs(bucket_name, prefix=prefix) return [blob.name for blob in blobs] def copy_blob(bucket_name: str, blob_name: str, destination_bucket: str, destination_blob: str): """Copy a blob within or across buckets.""" storage_client = storage.Client() source_bucket = storage_client.bucket(bucket_name) source_blob = source_bucket.blob(blob_name) dest_bucket = storage_client.bucket(destination_bucket) source_bucket.copy_blob(source_blob, dest_bucket, destination_blob) print(f"Blob {blob_name} copied to {destination_blob}") ``` #### Signed URLs & Lifecycle (09-gcs-signed-urls-lifecycle) ```python from google.cloud import storage from datetime import timedelta def generate_signed_url(bucket_name: str, blob_name: str, expiration_minutes: int = 15): """Generate a v4 signed URL for secure access.""" storage_client = storage.Client() bucket = storage_client.bucket(bucket_name) blob = bucket.blob(blob_name) url = blob.generate_signed_url( version="v4", expiration=timedelta(minutes=expiration_minutes), method="GET" ) return url def set_lifecycle_policy(bucket_name: str): """Set lifecycle rules for storage cost optimization.""" storage_client = storage.Client() bucket = storage_client.bucket(bucket_name) lifecycle_rules = [ { "action": {"type": "SetStorageClass", "storageClass": "NEARLINE"}, "condition": {"age": 30, "matchesPrefix": ["archive/"]} }, { "action": {"type": "Delete"}, "condition": {"age": 365, "matchesPrefix": ["temp/"]} } ] bucket.lifecycle_rules = lifecycle_rules bucket.patch() print(f"Lifecycle policy set for bucket {bucket_name}") ``` ### Pub/Sub Streaming (31-pubsub-streaming-pipeline) ```python from google.cloud import pubsub_v1 import json def publish_messages(project_id: str, topic_name: str, messages: list): """Publish messages to Pub/Sub topic.""" publisher = pubsub_v1.PublisherClient() topic_path = publisher.topic_path(project_id, topic_name) futures = [] for message in messages: message_json = json.dumps(message) future = publisher.publish( topic_path, message_json.encode("utf-8"), origin="data-pipeline", priority="high" ) futures.append(future) # Wait for all messages to publish for future in futures: future.result() print(f"Published {len(messages)} messages to {topic_name}") def subscribe_messages(project_id: str, subscription_name: str, callback_fn, timeout: int = None): """Subscribe and process messages from Pub/Sub.""" subscriber = pubsub_v1.SubscriberClient() subscription_path = subscriber.subscription_path( project_id, subscription_name ) flow_control = pubsub_v1.types.FlowControl( max_messages=100, max_bytes=10 * 1024 * 1024, # 10MB ) streaming_pull_future = subscriber.subscribe( subscription_path, callback=callback_fn, flow_control=flow_control ) print(f"Listening for messages on {subscription_path}...") try: streaming_pull_future.result(timeout=timeout) except KeyboardInterrupt: streaming_pull_future.cancel() # Example callback def message_callback(message): """Process received message.""" print(f"Received: {message.data.decode('utf-8')}") message.ack() ``` ### Apache Beam / Dataflow (47-50) #### Basic Beam Pipeline (47-beam-data-transformation) ```python import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions def run_word_count_pipeline(input_path: str, output_path: str): """Classic WordCount example with Beam.""" options = PipelineOptions() with beam.Pipeline(options=options) as pipeline: (pipeline | 'Read' >> beam.io.ReadFromText(input_path) | 'Split' >> beam.FlatMap(lambda line: line.split()) | 'PairWithOne' >> beam.Map(lambda word: (word, 1)) | 'GroupAndSum' >> beam.CombinePerKey(sum) | 'Format' >> beam.Map(lambda kv: f"{kv[0]}: {kv[1]}") | 'Write' >> beam.io.WriteToText(output_path) ) def csv_transform_pipeline(input_file: str, output_file: str): """Transform CSV data with Beam.""" def parse_csv(line): import csv from io import StringIO reader = csv.DictReader(StringIO(line)) return next(reader) def transform_record(record): return { 'id': record['id'], 'name': record['name'].upper(), 'value': float(record['value']) * 1.1, 'processed': True } options = PipelineOptions() with beam.Pipeline(options=options) as pipeline: (pipeline | 'Read CSV' >> beam.io.ReadFromText(input_file, skip_header_lines=1) | 'Parse' >> beam.Map(parse_csv) | 'Transform' >> beam.Map(transform_record) | 'Format JSON' >> beam.Map(lambda x: json.dumps(x)) | 'Write' >> beam.io.WriteToText(output_file) ) ``` #### Beam to BigQuery (48-beam-csv-to-bigquery-load) ```python import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.io.gcp.bigquery import WriteToBigQuery def csv_to_bigquery_pipeline( input_file: str, project_id: str, dataset_id: str, table_id: str ): """Load CSV to BigQuery using Beam.""" table_spec = f"{project_id}:{dataset_id}.{table_id}" table_schema = { 'fields': [ {'name': 'id', 'type': 'INTEGER', 'mode': 'REQUIRED'}, {'name': 'name', 'type': 'STRING', 'mode': 'REQUIRED'}, {'name': 'value', 'type': 'FLOAT', 'mode': 'NULLABLE'}, {'name': 'timestamp', 'type': 'TIMESTAMP', 'mode': 'REQUIRED'} ] } def parse_csv_row(line): parts = line.split(',') return { 'id': int(parts[0]), 'name': parts[1], 'value': float(parts[2]), 'timestamp': parts[3] } options = PipelineOptions() with beam.Pipeline(options=options) as pipeline: (pipeline