google-cloud-data-engineering-hub
Production-grade GCP data engineering projects using BigQuery, Dataflow, Beam, Composer, Pub/Sub, Dataproc, and Vertex AI
ソース情報
- リポジトリ
- reason-machines/data-skills
- ソースの最終更新活動
- 2026年6月13日 20:38
- 検出された SKILL.md の言語
- 英語
- スター
- 5
- フォーク
- 1
インストール方法
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
ソースファイルを確認
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
SKILL.md を表示中
SKILL.md
ソースの指示 · 読み取り専用プレビュー- name
- google-cloud-data-engineering-hub
- description
- Production-grade GCP data engineering projects using BigQuery, Dataflow, Beam, Composer, Pub/Sub, Dataproc, and Vertex AI
- triggers
- ["build a GCP data pipeline with BigQuery","set up Google Cloud data engineering project","create Dataflow pipeline on GCP","use Apache Beam with BigQuery","implement Cloud Composer workflow","process data with Dataproc Serverless","build streaming pipeline with Pub/Sub","integrate Gemini AI with data pipeline"]
# 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
GitHubで見るこの SKILL.md は非常に大きいため、SkillsMP では最初のセクションだけを表示しています。 GitHubで見る
