Cloud Big Data Services

# AWS CLI: Upload data to S3
aws s3 cp large_dataset.parquet s3://my-data-lake/raw-data/

# Query S3 directly with Athena (serverless SQL)
SELECT customer_id, SUM(revenue) as total
FROM "s3://my-data-lake/sales/*.parquet"
WHERE year = 2024
GROUP BY customer_id
LIMIT 100;

# Launch 50-node Spark cluster
aws emr create-cluster \
  --name "ETL Pipeline" \
  --release-label emr-6.15.0 \
  --applications Name=Spark Name=Hadoop \
  --instance-type m5.xlarge \
  --instance-count 50 \
  --use-default-roles

# Submit Spark job
aws emr add-steps \
  --cluster-id j-XXXXXXXXXXXXX \
  --steps Type=Spark,Name="Process Data",\
    Args=[--class,com.example.ETL,s3://my-bucket/etl.jar]

-- Load data from S3
COPY sales FROM 's3://my-bucket/sales/'
IAM_ROLE 'arn:aws:iam::xxx:role/RedshiftRole'
FORMAT AS PARQUET;

-- Query 10 billion rows (sub-second)
SELECT product_category, 
       SUM(revenue) as total_revenue,
       COUNT(DISTINCT customer_id) as customers
FROM sales
WHERE order_date >= '2024-01-01'
GROUP BY product_category
ORDER BY total_revenue DESC
LIMIT 10;

-- Create table pointing to S3 Parquet files
CREATE EXTERNAL TABLE logs (
  user_id STRING,
  timestamp TIMESTAMP,
  action STRING,
  page_url STRING
)
STORED AS PARQUET
LOCATION 's3://my-bucket/logs/year=2024/';

-- Query directly (no data loading!)
SELECT action, COUNT(*) as count
FROM logs
WHERE timestamp >= TIMESTAMP '2024-01-01'
GROUP BY action
ORDER BY count DESC;

# Python: Send data to Kinesis
import json
import boto3
kinesis = boto3.client('kinesis')

kinesis.put_record(
    StreamName='clickstream',
    Data=json.dumps({
        'user_id': '12345',
        'action': 'purchase',
        'amount': 99.99
    }),
    PartitionKey='user_12345'
)

# Process with Lambda (serverless)
# Triggered automatically for each record

# Azure CLI: Upload to Blob Storage
az storage blob upload-batch \
  --account-name mydatalake \
  --destination raw-data \
  --source ./local-data/ \
  --pattern "*.parquet"

# Enable hierarchical namespace (Data Lake Gen2)
az storage account update \
  --name mydatalake \
  --enable-hierarchical-namespace true

-- Serverless SQL pool (no infrastructure)
SELECT 
    product_category,
    SUM(revenue) as total_revenue
FROM OPENROWSET(
    BULK 'https://mydatalake.dfs.core.windows.net/sales/*.parquet',
    FORMAT = 'PARQUET'
) AS sales
WHERE order_date >= '2024-01-01'
GROUP BY product_category;

-- Also run Spark notebooks in same workspace
spark.read.parquet("abfss://sales@mydatalake.dfs.core.windows.net/")
     .groupBy("category")
     .agg(sum("revenue"))
     .show()

# Azure CLI: Create Spark cluster
az hdinsight create \
  --name spark-cluster \
  --resource-group myResourceGroup \
  --type spark \
  --cluster-tier Standard \
  --component-version Spark=3.1 \
  --headnode-size Standard_D12_v2 \
  --workernode-count 10 \
  --workernode-size Standard_D13_v2

# Python: Send to Event Hub (Kafka protocol)
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers='myeventhub.servicebus.windows.net:9093',
    security_protocol='SASL_SSL',
    sasl_mechanism='PLAIN',
    sasl_plain_username='$ConnectionString',
    sasl_plain_password='Endpoint=sb://...'
)

producer.send('telemetry', b'sensor data')
producer.flush()

# gsutil: Upload to Cloud Storage
gsutil -m cp -r ./local-data gs://my-data-lake/raw-data/

# Create external table for GCS data, then query
CREATE EXTERNAL TABLE `project.dataset.sales_external`
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://my-data-lake/sales/*.parquet']
);

SELECT customer_id, SUM(revenue) as total
FROM `project.dataset.sales_external`
GROUP BY customer_id;

-- Load data from Cloud Storage
LOAD DATA INTO `project.dataset.sales`
FROM FILES (
  format = 'PARQUET',
  uris = ['gs://my-bucket/sales/*.parquet']
) WITH CONNECTION `project.region.connection_id`;

-- Query 10 billion rows (no cluster to manage!)
SELECT 
  DATE_TRUNC(order_date, MONTH) as month,
  product_category,
  SUM(revenue) as revenue,
  APPROX_COUNT_DISTINCT(customer_id) as customers
FROM `project.dataset.sales`
WHERE order_date >= '2024-01-01'
GROUP BY month, product_category
ORDER BY revenue DESC;

# Create ephemeral cluster (auto-deletes after job)
gcloud dataproc jobs submit pyspark \
  --cluster-name=temp-cluster \
  --region=us-central1 \
  --cluster-labels=ephemeral=true \
  gs://my-bucket/etl-job.py

# Use preemptible workers (80% cheaper)
gcloud dataproc clusters create my-cluster \
  --num-workers=50 \
  --num-preemptible-workers=100 \
  --worker-machine-type=n1-standard-4

# Python: Apache Beam pipeline
import json
import apache_beam as beam

with beam.Pipeline() as pipeline:
    (pipeline
        | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub('projects/my-project/topics/events')
        | 'Parse JSON' >> beam.Map(json.loads)
        | 'Extract fields' >> beam.Map(lambda x: (x['user_id'], x['revenue']))
        | 'Sum by user' >> beam.CombinePerKey(sum)
        | 'Write to BigQuery' >> beam.io.WriteToBigQuery('project:dataset.user_revenue')
    )

# Python: Publish to Pub/Sub
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('my-project', 'events')

# Publish message
future = publisher.publish(
    topic_path,
    b'{"user_id": "123", "action": "purchase"}',
    event_type='purchase'
)
print(f"Published message ID: {future.result()}")