import boto3

s3 = boto3.client("s3")

# An object's "path" is really just its key string
bucket = "my-company-data-lake"
key    = "raw/sales/2024/01/15/transactions.parquet"

# Upload a local file as an object
s3.upload_file("transactions.parquet", bucket, key)

# s3://my-company-data-lake/raw/sales/2024/01/15/transactions.parquet
print(f"s3://{bucket}/{key}")

response = s3.list_objects_v2(
    Bucket="my-company-data-lake",
    Prefix="raw/sales/2024/01/",   # acts like a "folder"
    Delimiter="/"                  # stops at the next "folder" level
)

# CommonPrefixes shows the "sub-folders" (day=01, day=02, ...)
for p in response.get("CommonPrefixes", []):
    print(p["Prefix"])

# Custom metadata is set at upload time
s3.put_object(
    Bucket=bucket, Key=key, Body=open("data.parquet","rb"),
    Metadata={"source-system": "salesforce", "pipeline": "sales-etl"}
)

# Tags can be added/changed anytime — used for cost allocation & lifecycle
s3.put_object_tagging(
    Bucket=bucket, Key=key,
    Tagging={"TagSet": [
        {"Key": "environment", "Value": "production"},
        {"Key": "team", "Value": "data-engineering"}
    ]}
)

s3.put_bucket_lifecycle_configuration(
    Bucket="my-company-data-lake",
    LifecycleConfiguration={
        "Rules": [{
            "ID": "raw-zone-tiering",
            "Filter": {"Prefix": "raw/"},
            "Status": "Enabled",
            "Transitions": [
                {"Days": 30,  "StorageClass": "STANDARD_IA"},
                {"Days": 365, "StorageClass": "GLACIER_DEEP_ARCHIVE"}
            ],
            "Expiration": {"Days": 2555}  # ~7 years
        }]
    }
)

iam = boto3.client("iam")

# Managed policy — reusable, attach by ARN
iam.attach_role_policy(
    RoleName="glue-etl-role",
    PolicyArn="arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess"
)

# Inline policy — embedded directly in this one role only
iam.put_role_policy(
    RoleName="glue-etl-role",
    PolicyName="AllowWriteToSilverBucket",
    PolicyDocument=json.dumps({
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Action": ["s3:PutObject", "s3:DeleteObject"],
            "Resource": "arn:aws:s3:::company-silver/*"
        }]
    })
)

import boto3

kms = boto3.client("kms", region_name="us-east-1")

# Create a symmetric CMK for encrypting data lake data
response = kms.create_key(
    Description="CMK for production data lake encryption",
    KeyUsage="ENCRYPT_DECRYPT",       # default — symmetric AES-256 key
    KeySpec="SYMMETRIC_DEFAULT",        # AES-256 GCM
    Policy="""
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "Enable IAM Root",
          "Effect": "Allow",
          "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
          "Action": "kms:*",
          "Resource": "*"
        },
        {
          "Sid": "Allow Glue and EMR roles to use the key",
          "Effect": "Allow",
          "Principal": {
            "AWS": [
              "arn:aws:iam::123456789012:role/GlueExecutionRole",
              "arn:aws:iam::123456789012:role/EMRJobRole"
            ]
          },
          "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
          "Resource": "*"
        }
      ]
    }
    """
)

key_id  = response["KeyMetadata"]["KeyId"]
key_arn = response["KeyMetadata"]["Arn"]
print(f"Created CMK: {key_arn}")

# Create a human-readable alias so you don't have to remember the UUID
kms.create_alias(
    AliasName="alias/data-lake-cmk",
    TargetKeyId=key_id
)
# Now you can reference it as alias/data-lake-cmk in all service configs

import boto3
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

kms    = boto3.client("kms")
CMK_ID = "alias/data-lake-cmk"

# ─── ENCRYPT ───────────────────────────────────────────────
# Step 1: Ask KMS for a data key
dk_resp = kms.generate_data_key(KeyId=CMK_ID, KeySpec="AES_256")

plaintext_data_key  = dk_resp["Plaintext"]      # 32 bytes — USE then DESTROY
encrypted_data_key  = dk_resp["CiphertextBlob"] # store alongside your data

# Step 2: Encrypt your actual data LOCALLY (fast, no KMS call)
data_to_encrypt = b"sensitive customer data here"
nonce = os.urandom(12)                            # 96-bit nonce for AES-GCM
aesgcm = AESGCM(plaintext_data_key)
ciphertext = aesgcm.encrypt(nonce, data_to_encrypt, None)

# Step 3: Clear the plaintext key from memory
del plaintext_data_key
# Store: ciphertext + nonce + encrypted_data_key together

# ─── DECRYPT ───────────────────────────────────────────────
# Step 1: Call KMS to decrypt the encrypted data key
dk_plain = kms.decrypt(CiphertextBlob=encrypted_data_key)["Plaintext"]

# Step 2: Decrypt locally
aesgcm2    = AESGCM(dk_plain)
decrypted  = aesgcm2.decrypt(nonce, ciphertext, None)
print(decrypted)  # b"sensitive customer data here"

# Step 3: Clear plaintext key
del dk_plain
# Note: In practice S3/Glue/Redshift do ALL of this for you automatically!

import boto3

s3     = boto3.client("s3")
CMK_ID = "arn:aws:kms:us-east-1:123456789012:alias/data-lake-cmk"

# Upload with SSE-KMS using your CMK
s3.upload_file(
    Filename="sales_data.parquet",
    Bucket="my-data-lake",
    Key="gold/sales/2024/sales_data.parquet",
    ExtraArgs={
        "ServerSideEncryption": "aws:kms",
        "SSEKMSKeyId": CMK_ID
    }
)

# Alternatively via put_object
s3.put_object(
    Bucket="my-data-lake",
    Key="gold/users/users.parquet",
    Body=b"parquet bytes here",
    ServerSideEncryption="aws:kms",
    SSEKMSKeyId=CMK_ID
)

import json, boto3

s3     = boto3.client("s3")
BUCKET = "my-data-lake"
CMK_ARN = "arn:aws:kms:us-east-1:123456789012:key/abcd-1234"

# Bucket policy that DENIES any upload without SSE-KMS using our CMK
policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DenyNonKMSUploads",
            "Effect": "Deny",
            "Principal": "*",
            "Action": "s3:PutObject",
            "Resource": f"arn:aws:s3:::{BUCKET}/*",
            "Condition": {
                "StringNotEquals": {
                    "s3:x-amz-server-side-encryption": "aws:kms"
                }
            }
        },
        {
            "Sid": "DenyWrongKMSKey",
            "Effect": "Deny",
            "Principal": "*",
            "Action": "s3:PutObject",
            "Resource": f"arn:aws:s3:::{BUCKET}/*",
            "Condition": {
                "StringNotEquals": {
                    "s3:x-amz-server-side-encryption-aws-kms-key-id": CMK_ARN
                }
            }
        }
    ]
}

s3.put_bucket_policy(
    Bucket=BUCKET,
    Policy=json.dumps(policy)
)
print("Bucket policy applied — unencrypted uploads will now be denied.")

import boto3

glue   = boto3.client("glue")
CMK_ARN = "arn:aws:kms:us-east-1:123456789012:alias/data-lake-cmk"

# Create a security configuration
glue.create_security_configuration(
    Name="data-lake-security-config",
    EncryptionConfiguration={
        "S3Encryption": [
            {
                "S3EncryptionMode": "SSE-KMS",     # encrypt all S3 output
                "KmsKeyArn": CMK_ARN
            }
        ],
        "CloudWatchEncryption": {
            "CloudWatchEncryptionMode": "SSE-KMS",  # encrypt job logs
            "KmsKeyArn": CMK_ARN
        },
        "JobBookmarksEncryption": {
            "JobBookmarksEncryptionMode": "CSE-KMS", # encrypt incremental state
            "KmsKeyArn": CMK_ARN
        }
    }
)

# Attach security configuration when creating / updating a Glue job
glue.create_job(
    Name="sales-etl-job",
    Role="arn:aws:iam::123456789012:role/GlueExecutionRole",
    Command={"Name": "glueetl", "ScriptLocation": "s3://scripts/etl.py", "PythonVersion": "3"},
    SecurityConfiguration="data-lake-security-config",  # ← attach here
    GlueVersion="4.0",
    NumberOfWorkers=10,
    WorkerType="G.1X"
)

import boto3

redshift = boto3.client("redshift")
CMK_ARN  = "arn:aws:kms:us-east-1:123456789012:alias/data-lake-cmk"

redshift.create_cluster(
    ClusterIdentifier="prod-dwh",
    NodeType="ra3.4xlarge",
    NumberOfNodes=4,
    MasterUsername="admin",
    MasterUserPassword="ChangeMe123!",  # use Secrets Manager in prod!
    DBName="analytics",
    Encrypted=True,           # enable encryption
    KmsKeyId=CMK_ARN,          # use our CMK — not AWS managed key
    VpcSecurityGroupIds=["sg-abc123"],
    ClusterSubnetGroupName="redshift-subnet-group"
)
print("Encrypted Redshift cluster creation started.")

import boto3, json

sm      = boto3.client("secretsmanager")
CMK_ARN = "arn:aws:kms:us-east-1:123456789012:alias/data-lake-cmk"

# Create a secret encrypted with our CMK
sm.create_secret(
    Name="prod/postgresql/credentials",
    Description="PostgreSQL DB credentials for ETL pipeline",
    KmsKeyId=CMK_ARN,   # ← associate our CMK here
    SecretString=json.dumps({
        "host":     "prod-db.cluster-xyz.us-east-1.rds.amazonaws.com",
        "port":     5432,
        "username": "etl_user",
        "password": "SuperSecret123!",
        "dbname":   "analytics"
    })
)
print("Secret stored with CMK encryption.")

import boto3

kms    = boto3.client("kms")
KEY_ID = "alias/data-lake-cmk"

# Enable automatic rotation (rotates every 365 days by default)
kms.enable_key_rotation(
    KeyId=KEY_ID,
    RotationPeriodInDays=365   # annual rotation (90–2560 days)
)

# Check rotation status
status = kms.get_key_rotation_status(KeyId=KEY_ID)
print(status["KeyRotationEnabled"])          # True
print(status.get("NextRotationDate"))        # e.g. 2025-01-15T00:00:00Z

# You can also trigger a manual on-demand rotation right now
kms.rotate_key_on_demand(KeyId=KEY_ID)
print("On-demand rotation triggered.")

import boto3

kms = boto3.client("kms")

# Step 1: Create a brand new CMK
new_key = kms.create_key(
    Description="Data lake CMK v2 — 2025 rotation",
    KeyUsage="ENCRYPT_DECRYPT",
    KeySpec="SYMMETRIC_DEFAULT"
)
new_key_id = new_key["KeyMetadata"]["KeyId"]

# Step 2: Move the alias to the new key
# (all configs referencing "alias/data-lake-cmk" now use the new key)
kms.update_alias(
    AliasName="alias/data-lake-cmk",
    TargetKeyId=new_key_id
)
print(f"Alias now points to new key: {new_key_id}")
# Old data encrypted with old key is still readable.
# New data is encrypted with new_key_id.
# Old key should be scheduled for deletion after data migration.

# Step 3: Schedule old key for deletion (7–30 day waiting period)
OLD_KEY_ID = "mrk-previous-key-id"
kms.schedule_key_deletion(
    KeyId=OLD_KEY_ID,
    PendingWindowInDays=30   # 30-day grace period to recover if needed
)

import boto3

kms = boto3.client("kms")

# List all CMKs and their aliases
for page in kms.get_paginator("list_keys").paginate():
    for key in page["Keys"]:
        detail = kms.describe_key(KeyId=key["KeyId"])["KeyMetadata"]
        if detail["KeyManager"] == "CUSTOMER":   # only our CMKs
            print(detail["KeyId"], detail["Description"], detail["KeyState"])

# Disable a key (emergency — data becomes unreadable)
kms.disable_key(KeyId="alias/data-lake-cmk")

# Re-enable a key
kms.enable_key(KeyId="alias/data-lake-cmk")

# Check if a key policy allows your Glue role to use it
policy = kms.get_key_policy(
    KeyId="alias/data-lake-cmk",
    PolicyName="default"
)["Policy"]
print(policy)   # JSON string — inspect for your role ARN

import boto3

s3     = boto3.client("s3")
BUCKET = "my-data-lake"
CMK_ARN = "arn:aws:kms:us-east-1:123456789012:alias/data-lake-cmk"

# Enable default SSE-KMS with Bucket Key on the bucket
s3.put_bucket_encryption(
    Bucket=BUCKET,
    ServerSideEncryptionConfiguration={
        "Rules": [{
            "ApplyServerSideEncryptionByDefault": {
                "SSEAlgorithm": "aws:kms",
                "KMSMasterKeyID": CMK_ARN
            },
            "BucketKeyEnabled": True   # ← this is the cost saver
        }]
    }
)
print("Bucket Key enabled — KMS API costs reduced by ~99%.")

import boto3, json

sm = boto3.client("secretsmanager", region_name="us-east-1")

# Store a database credential as JSON — the standard pattern
secret_value = json.dumps({
    "username": "pipeline_user",
    "password": "S3cr3tP@ssw0rd!",
    "host":     "prod-rds.cluster-abc.us-east-1.rds.amazonaws.com",
    "port":     5432,
    "dbname":  "analytics"
})

response = sm.create_secret(
    Name="prod/rds/pipeline-user",          # hierarchical name
    Description="RDS credentials for the nightly ETL pipeline",
    SecretString=secret_value,
    KmsKeyId="alias/data-lake-cmk",          # encrypt with CMK (not default key)
    Tags=[
        {"Key": "environment", "Value": "production"},
        {"Key": "team",        "Value": "data-engineering"}
    ]
)

print(f"Secret ARN: {response['ARN']}")
# arn:aws:secretsmanager:us-east-1:123456789012:secret:prod/rds/pipeline-user-AbCdEf

import boto3, json
from botocore.exceptions import ClientError

def get_secret(secret_name: str, region: str = "us-east-1") -> dict:
    """Retrieve a JSON secret from AWS Secrets Manager."""
    sm = boto3.client("secretsmanager", region_name=region)
    try:
        response = sm.get_secret_value(SecretId=secret_name)
    except ClientError as e:
        code = e.response["Error"]["Code"]
        if code == "ResourceNotFoundException":
            raise ValueError(f"Secret '{secret_name}' does not exist.")
        elif code == "AccessDeniedException":
            raise PermissionError(f"IAM role lacks permission to read '{secret_name}'.")
        else:
            raise   # re-raise unexpected errors

    # SecretString for text secrets (most common), SecretBinary for binary
    return json.loads(response["SecretString"])


# ── Usage — RDS connection ───────────────────────────────────
creds = get_secret("prod/rds/pipeline-user")

# creds is now a plain dict — unpack what you need
DB_HOST = creds["host"]
DB_PORT = creds["port"]
DB_USER = creds["username"]
DB_PASS = creds["password"]
DB_NAME = creds["dbname"]

# Use them to build a JDBC URL for Spark
jdbc_url = f"jdbc:postgresql://{DB_HOST}:{DB_PORT}/{DB_NAME}"

df = spark.read \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("user", DB_USER) \
    .option("password", DB_PASS) \
    .option("dbtable", "public.orders") \
    .load()

# glue_etl_job.py  — this script runs on a Glue executor
import sys, json, boto3
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc   = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# ── Fetch Snowflake creds from Secrets Manager at runtime ──
sm = boto3.client("secretsmanager")
sf_creds = json.loads(
    sm.get_secret_value(SecretId="prod/snowflake/etl-role")["SecretString"]
)

# ── Use them in the Spark Snowflake connector ──
sf_options = {
    "sfURL":      sf_creds["sfURL"],
    "sfUser":     sf_creds["sfUser"],
    "sfPassword": sf_creds["sfPassword"],
    "sfDatabase": "ANALYTICS",
    "sfWarehouse":"ETL_WH",
    "sfSchema":   "SILVER",
}

df = spark.read \
    .format("net.snowflake.spark.snowflake") \
    .options(**sf_options) \
    .option("dbtable", "orders") \
    .load()

# creds never appear in logs, git, or job parameters — they are fetched live

import json, boto3

sm = boto3.client("secretsmanager")

# ── Module-level cache — fetched once per container warm-start ──
_DB_CREDS = None

def get_db_creds():
    global _DB_CREDS
    if _DB_CREDS is None:
        raw = sm.get_secret_value(SecretId="prod/rds/pipeline-user")
        _DB_CREDS = json.loads(raw["SecretString"])
    return _DB_CREDS

def lambda_handler(event, context):
    creds = get_db_creds()   # uses cached value after first call
    # ... use creds to connect to RDS and process the event ...
    return {"statusCode": 200, "body": "OK"}

# First invocation: hits Secrets Manager API (~5ms latency)
# Subsequent invocations on same container: reads from _DB_CREDS (0ms)

# In airflow.cfg or MWAA environment variables:
# [secrets]
# backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
# backend_kwargs = {"connections_prefix": "airflow/connections", "variables_prefix": "airflow/variables"}

# Then a connection stored as:  airflow/connections/my_rds_conn
# and a variable stored as:     airflow/variables/my_s3_bucket
# are automatically available in DAGs as:
#   BaseHook.get_connection("my_rds_conn")
#   Variable.get("my_s3_bucket")

# Manual retrieval inside a DAG task (if not using the backend):
import boto3, json

def my_task(**context):
    sm = boto3.client("secretsmanager")
    creds = json.loads(
        sm.get_secret_value(SecretId="prod/rds/pipeline-user")["SecretString"]
    )
    # build JDBC URL and run Spark job...

import boto3

sm = boto3.client("secretsmanager")

# Enable rotation — AWS has built-in Lambda functions for RDS, Redshift, etc.
sm.rotate_secret(
    SecretId="prod/rds/pipeline-user",
    RotationLambdaARN="arn:aws:lambda:us-east-1:123456789012:function:SecretsManagerRDSRotation",
    RotationRules={
        "AutomaticallyAfterDays": 30   # rotate every 30 days
    }
)

# For supported services (RDS, Aurora, Redshift, DocumentDB), AWS provides
# pre-built rotation Lambda functions in the Serverless Application Repository.
# For custom services (Snowflake, APIs), you write your own rotation Lambda.

# Check rotation status
secret_meta = sm.describe_secret(SecretId="prod/rds/pipeline-user")
print("Rotation enabled:", secret_meta.get("RotationEnabled"))
print("Last rotated:",    secret_meta.get("LastRotatedDate"))
print("Next rotation:",   secret_meta.get("NextRotationDate"))

import boto3, json

sm = boto3.client("secretsmanager")

# Default: fetches AWSCURRENT (latest valid secret)
current = json.loads(sm.get_secret_value(SecretId="prod/rds/pipeline-user")["SecretString"])

# Explicitly fetch the previous version (rollback scenario)
previous = json.loads(sm.get_secret_value(
    SecretId="prod/rds/pipeline-user",
    VersionStage="AWSPREVIOUS"
)["SecretString"])

# List all versions of a secret to see the labels
versions = sm.list_secret_version_ids(SecretId="prod/rds/pipeline-user")
for v in versions["Versions"]:
    print(v["VersionId"], v.get("VersionStages"))
# e.g:
# abc123  ['AWSCURRENT']
# def456  ['AWSPREVIOUS']

import boto3, json, string, secrets

sm = boto3.client("secretsmanager")

def lambda_handler(event, context):
    step       = event["Step"]
    secret_id  = event["SecretId"]
    token      = event["ClientRequestToken"]   # new version ID

    if step == "createSecret":
        # Generate a new password and store as AWSPENDING version
        new_pass = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(32))
        current  = json.loads(sm.get_secret_value(SecretId=secret_id)["SecretString"])
        current["password"] = new_pass
        sm.put_secret_value(
            SecretId=secret_id, ClientRequestToken=token,
            SecretString=json.dumps(current), VersionStages=["AWSPENDING"]
        )

    elif step == "setSecret":
        # Apply the new password to the actual service (e.g. Snowflake ALTER USER)
        pending = json.loads(sm.get_secret_value(
            SecretId=secret_id, VersionStage="AWSPENDING")["SecretString"])
        # ... call Snowflake / RDS / API to set the new password ...

    elif step == "testSecret":
        # Verify the pending credentials actually work
        pending = json.loads(sm.get_secret_value(
            SecretId=secret_id, VersionStage="AWSPENDING")["SecretString"])
        # ... try connecting with pending creds; raise exception if it fails ...

    elif step == "finishSecret":
        # Promote AWSPENDING → AWSCURRENT
        sm.update_secret_version_stage(
            SecretId=secret_id, VersionStage="AWSCURRENT",
            MoveToVersionId=token,
            RemoveFromVersionId=sm.describe_secret(SecretId=secret_id)["VersionIdsToStages"]
                             and None  # simplified for illustration
        )

import boto3, json

sm = boto3.client("secretsmanager", region_name="us-east-1")

# ── 1. CREATE a secret ───────────────────────────────────────
sm.create_secret(
    Name="prod/kafka/sasl-credentials",
    SecretString=json.dumps({"username": "kafka-user", "password": "s3cr3t"}),
    KmsKeyId="alias/data-lake-cmk"
)

# ── 2. GET a secret (the most common call) ───────────────────
creds = json.loads(
    sm.get_secret_value(SecretId="prod/kafka/sasl-credentials")["SecretString"]
)

# ── 3. UPDATE (rotate/change) a secret value ─────────────────
sm.put_secret_value(
    SecretId="prod/kafka/sasl-credentials",
    SecretString=json.dumps({"username": "kafka-user", "password": "n3wP@ss!"})
)

# ── 4. DESCRIBE a secret (metadata, rotation status) ─────────
meta = sm.describe_secret(SecretId="prod/kafka/sasl-credentials")
print(meta["RotationEnabled"])   # False / True
print(meta["LastChangedDate"])   # datetime
print(meta["Tags"])              # list of {Key, Value}

# ── 5. LIST secrets (with paginator) ─────────────────────────
paginator = sm.get_paginator("list_secrets")
for page in paginator.paginate(
    Filters=[{"Key": "tag-key", "Values": ["environment"]}]
):
    for secret in page["SecretList"]:
        print(secret["Name"], secret["LastChangedDate"])

# ── 6. DELETE a secret ──────────────────────────────────────
sm.delete_secret(
    SecretId="prod/kafka/sasl-credentials",
    RecoveryWindowInDays=30   # 30-day window before permanent deletion
    # ForceDeleteWithoutRecovery=True  ← immediate deletion (no recovery!)
)

# ── 7. RESTORE a deleted secret (within recovery window) ─────
sm.restore_secret(SecretId="prod/kafka/sasl-credentials")

# ── 8. TAG a secret ─────────────────────────────────────────
sm.tag_resource(
    SecretId="prod/kafka/sasl-credentials",
    Tags=[{"Key": "cost-center", "Value": "data-platform"}]
)

# ── 9. REPLICATE to another region (DR pattern) ──────────────
sm.replicate_secret_to_regions(
    SecretId="prod/rds/pipeline-user",
    AddReplicaRegions=[{"Region": "us-west-2", "KmsKeyId": "alias/data-lake-cmk-west"}]
)
# Now your pipelines in us-west-2 read from the local replica — faster & resilient

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue",
        "secretsmanager:DescribeSecret"
      ],
      "Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:prod/sales/*"
    },
    {
      "Effect": "Allow",
      "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
      "Resource": "arn:aws:kms:us-east-1:123456789012:key/alias/data-lake-cmk"
    }
  ]
}

import boto3

ssm = boto3.client("ssm", region_name="us-east-1")

# ── String: plain non-sensitive config ──────────────────────────
ssm.put_parameter(
    Name="/datalake/prod/s3_bucket",
    Value="my-company-data-lake-prod",
    Type="String",
    Description="Main data lake S3 bucket for production",
    Overwrite=True
)

# ── StringList: comma-separated list of values ──────────────────
ssm.put_parameter(
    Name="/datalake/prod/active_regions",
    Value="us-east-1,eu-west-1,ap-southeast-1",
    Type="StringList",
    Description="Regions where the pipeline runs",
    Overwrite=True
)

# ── SecureString: encrypted with KMS (for mildly sensitive config)
ssm.put_parameter(
    Name="/datalake/prod/internal_api_key",
    Value="int-api-key-abc123",
    Type="SecureString",
    KeyId="alias/data-lake-cmk",   # optional: use CMK (default: AWS-managed key)
    Description="Internal monitoring API key",
    Overwrite=True
)

print("All parameters stored.")

import boto3

ssm = boto3.client("ssm", region_name="us-east-1")

# ✅ CORRECT — WithDecryption=True decrypts the SecureString
response = ssm.get_parameter(
    Name="/datalake/prod/internal_api_key",
    WithDecryption=True
)
api_key = response["Parameter"]["Value"]
print(api_key)   # "int-api-key-abc123" — actual value

# ❌ WRONG — without WithDecryption you get an encrypted blob
bad_response = ssm.get_parameter(
    Name="/datalake/prod/internal_api_key"
    # WithDecryption defaults to False!
)
print(bad_response["Parameter"]["Value"])
# AQICAHh7... (encrypted gibberish — DO NOT use this)

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadSSMParameters",
      "Effect": "Allow",
      "Action": [
        "ssm:GetParameter",
        "ssm:GetParameters",
        "ssm:GetParametersByPath"
      ],
      "Resource": "arn:aws:ssm:us-east-1:123456789012:parameter/datalake/prod/*"
    },
    {
      "Sid": "DecryptSSMSecureString",
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "arn:aws:kms:us-east-1:123456789012:key/alias/data-lake-cmk"
    }
  ]
}

import boto3

ssm = boto3.client("ssm", region_name="us-east-1")

def load_pipeline_config(path_prefix: str) -> dict:
    """Load all parameters under a path prefix as a flat dict."""
    config = {}
    paginator = ssm.get_paginator("get_parameters_by_path")

    for page in paginator.paginate(
        Path=path_prefix,
        Recursive=True,          # include all nested sub-paths
        WithDecryption=True     # decrypt SecureStrings automatically
    ):
        for param in page["Parameters"]:
            # Strip the path prefix to get the short key name
            short_name = param["Name"].replace(path_prefix, "").lstrip("/")
            config[short_name] = param["Value"]

    return config


# ── Usage in a Glue job ───────────────────────────────────────────
cfg = load_pipeline_config("/datalake/prod")

print(cfg)
# {
#   "s3_bucket":         "my-company-data-lake-prod",
#   "glue_database":     "prod_analytics",
#   "spark_partitions":  "200",
#   "dq_threshold_pct":  "95",
#   "kafka_topic":       "prod.sales.events",
#   "active_regions":    "us-east-1,eu-west-1,ap-southeast-1",
#   "internal_api_key":  "int-api-key-abc123"   ← SecureString, auto-decrypted
# }

# Now use config values naturally
s3_bucket  = cfg["s3_bucket"]
partitions = int(cfg["spark_partitions"])
dq_pct     = float(cfg["dq_threshold_pct"])
regions    = cfg["active_regions"].split(",")   # StringList → Python list

print(f"Writing to: {s3_bucket}")
print(f"Partitions: {partitions}, DQ threshold: {dq_pct}%")
print(f"Active regions: {regions}")

import boto3
from botocore.exceptions import ClientError

ssm = boto3.client("ssm")

def get_param(name: str, default=None):
    """Get a single parameter, returning default if it doesn't exist."""
    try:
        resp = ssm.get_parameter(Name=name, WithDecryption=True)
        return resp["Parameter"]["Value"]
    except ClientError as e:
        if e.response["Error"]["Code"] == "ParameterNotFound":
            return default
        raise   # re-raise unexpected errors


# ── Feature flag check in pipeline ───────────────────────────────
run_dq_checks = get_param("/datalake/prod/feature/run_dq_checks", default="true")

if run_dq_checks.lower() == "true":
    print("Running data quality checks...")
    # ... run checks ...
else:
    print("DQ checks disabled via feature flag — skipping.")

# ── Batch fetch multiple specific parameters ──────────────────────
resp = ssm.get_parameters(
    Names=[
        "/datalake/prod/s3_bucket",
        "/datalake/prod/glue_database",
        "/datalake/prod/kafka_topic"
    ],
    WithDecryption=True
)
params = {p["Name"].split("/")[-1]: p["Value"] for p in resp["Parameters"]}
# {"s3_bucket": "...", "glue_database": "...", "kafka_topic": "..."}

# Check for invalid/missing names
if resp.get("InvalidParameters"):
    print(f"WARNING: Parameters not found: {resp['InvalidParameters']}")

import boto3
from botocore.exceptions import ClientError

ssm = boto3.client("ssm", region_name="us-east-1")

# ── 1. CREATE or UPDATE a parameter (Overwrite=True = upsert) ────
ssm.put_parameter(
    Name="/datalake/prod/spark/shuffle_partitions",
    Value="200",
    Type="String",
    Description="Spark shuffle partition count for production",
    Overwrite=True
)

# ── 2. GET a single parameter ─────────────────────────────────────
resp  = ssm.get_parameter(
    Name="/datalake/prod/spark/shuffle_partitions",
    WithDecryption=True   # always set True; no-op for String type
)
value = resp["Parameter"]["Value"]        # "200"
ptype = resp["Parameter"]["Type"]         # "String"
ver   = resp["Parameter"]["Version"]      # version number (increments on update)

# ── 3. GET multiple specific parameters in one API call ───────────
multi = ssm.get_parameters(
    Names=[
        "/datalake/prod/s3/raw_bucket",
        "/datalake/prod/glue/database",
        "/datalake/prod/kafka/topic"
    ],
    WithDecryption=True
)
# Build a dict: strip full path, keep just the last key segment
params = {p["Name"].rsplit("/", 1)[-1]: p["Value"] for p in multi["Parameters"]}
# Check if any requested names were invalid/missing
if multi["InvalidParameters"]:
    raise ValueError(f"Missing parameters: {multi['InvalidParameters']}")

# ── 4. GET ALL parameters under a path (bulk config load) ─────────
paginator = ssm.get_paginator("get_parameters_by_path")
all_params = {}
for page in paginator.paginate(
    Path="/datalake/prod",
    Recursive=True,
    WithDecryption=True
):
    for p in page["Parameters"]:
        all_params[p["Name"]] = p["Value"]

# ── 5. GET a specific version of a parameter ──────────────────────
# Useful for rollback: fetch a known-good previous version
history_resp = ssm.get_parameter_history(
    Name="/datalake/prod/spark/shuffle_partitions",
    WithDecryption=True
)
for h in history_resp["Parameters"]:
    print(h["Version"], h["Value"], h["LastModifiedDate"])

# Fetch a specific version directly
v2 = ssm.get_parameter(
    Name="/datalake/prod/spark/shuffle_partitions:2",  # :version suffix
    WithDecryption=True
)["Parameter"]["Value"]

# ── 6. DELETE a parameter ─────────────────────────────────────────
ssm.delete_parameter(Name="/datalake/staging/temp_flag")

# Delete multiple at once (up to 10)
ssm.delete_parameters(
    Names=[
        "/datalake/staging/old_flag",
        "/datalake/staging/deprecated_key"
    ]
)

# ── 7. DESCRIBE (list parameters with metadata) ───────────────────
desc_paginator = ssm.get_paginator("describe_parameters")
for page in desc_paginator.paginate(
    ParameterFilters=[{
        "Key": "Path",
        "Option": "Recursive",
        "Values": ["/datalake/prod"]
    }]
):
    for p in page["Parameters"]:
        print(p["Name"], p["Type"], p.get("Description", ""))

# ── 8. ADD TAGS to a parameter ────────────────────────────────────
ssm.add_tags_to_resource(
    ResourceType="Parameter",
    ResourceId="/datalake/prod/spark/shuffle_partitions",
    Tags=[
        {"Key": "team",  "Value": "data-engineering"},
        {"Key": "owner", "Value": "pipeline-team"}
    ]
)

import sys, boto3, json
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext

# ── Step 1: Get job arguments (just the env) ──────────────────────
args = getResolvedOptions(sys.argv, ["JOB_NAME", "ENV"])
env  = args["ENV"]   # "prod" or "staging" — passed at job launch

# ── Step 2: Load ALL config for this environment in one call ──────
ssm = boto3.client("ssm")

def load_config(env: str) -> dict:
    paginator = ssm.get_paginator("get_parameters_by_path")
    cfg = {}
    for page in paginator.paginate(
        Path=f"/datalake/{env}",
        Recursive=True,
        WithDecryption=True
    ):
        for p in page["Parameters"]:
            short = p["Name"].replace(f"/datalake/{env}/", "")
            cfg[short] = p["Value"]
    return cfg

cfg = load_config(env)

# ── Step 3: Use config values in the job ─────────────────────────
RAW_BUCKET   = cfg["s3/raw_bucket"]
SILVER_BUCKET = cfg["s3/silver_bucket"]
GLUE_DB      = cfg["glue/database"]
PARTITIONS   = int(cfg["spark/shuffle_partitions"])
DQ_ENABLED   = cfg.get("feature/dq_enabled", "true") == "true"

print(f"[{env.upper()}] Loading from {RAW_BUCKET} → {SILVER_BUCKET}")
print(f"Shuffle partitions: {PARTITIONS}, DQ: {DQ_ENABLED}")

# ── Step 4: Spark job runs with config ────────────────────────────
sc    = SparkContext()
glue  = GlueContext(sc)
spark = glue.spark_session
spark.conf.set("spark.sql.shuffle.partitions", PARTITIONS)

df = spark.read.parquet(f"s3://{RAW_BUCKET}/bronze/sales/")
# ... transformations ...
df.write.mode("overwrite").parquet(f"s3://{SILVER_BUCKET}/silver/sales/")

import boto3, json

ssm = boto3.client("ssm")

# ── Advanced parameter with expiry in 90 days ────────────────────
# (requires Tier="Advanced")
ssm.put_parameter(
    Name="/datalake/prod/api/partner_token",
    Value="token-abc-xyz-123",
    Type="SecureString",
    Tier="Advanced",          # required for parameter policies
    Overwrite=True,
    Policies=json.dumps([
        {
            "Type": "Expiration",           # auto-delete after this date
            "Version": "1.0",
            "Attributes": {
                "Timestamp": "2025-12-31T23:59:59.000Z"
            }
        },
        {
            "Type": "ExpirationNotification",  # alert 14 days before expiry
            "Version": "1.0",
            "Attributes": {
                "Before":  "14",
                "Unit":    "Days"
            }
        },
        {
            "Type": "NoChangeNotification",     # alert if not updated for 30 days
            "Version": "1.0",
            "Attributes": {
                "After": "30",
                "Unit":  "Days"
            }
        }
    ])
)
print("Advanced parameter with policies created.")

import boto3

glue = boto3.client("glue", region_name="us-east-1")

# ── 1. Create a database ─────────────────────────────────────────
glue.create_database(DatabaseInput={
    "Name": "prod_analytics",
    "Description": "Production analytics — silver and gold tables",
    "LocationUri": "s3://my-data-lake/silver/"
})

# ── 2. Create an external Parquet table ──────────────────────────
glue.create_table(
    DatabaseName="prod_analytics",
    TableInput={
        "Name": "sales_transactions",
        "Description": "Cleaned and validated sales transactions",
        "StorageDescriptor": {
            "Columns": [
                {"Name": "txn_id",      "Type": "string"},
                {"Name": "customer_id",  "Type": "bigint"},
                {"Name": "amount",       "Type": "decimal(18,2)"},
                {"Name": "txn_date",     "Type": "date"},
                {"Name": "product_id",   "Type": "string"}
            ],
            "Location": "s3://my-data-lake/silver/sales/",
            "InputFormat":  "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
            "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
            "SerdeInfo": {
                "SerializationLibrary": "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
            },
            "Compressed": True
        },
        "PartitionKeys": [
            {"Name": "year",  "Type": "string"},
            {"Name": "month", "Type": "string"}
        ],
        "TableType": "EXTERNAL_TABLE"
    }
)
print("Table 'sales_transactions' registered in Glue Catalog.")

import boto3, copy

glue = boto3.client("glue")

# 1. Get the current table definition
resp  = glue.get_table(DatabaseName="prod_analytics", Name="sales_transactions")
table = resp["Table"]

# 2. Build the TableInput (must strip read-only fields AWS adds)
table_input = {
    k: v for k, v in table.items()
    if k not in ["DatabaseName", "CreateTime", "UpdateTime",
                  "CreatedBy", "IsRegisteredWithLakeFormation",
                  "CatalogId", "VersionId"]
}

# 3. Add the new column
table_input["StorageDescriptor"]["Columns"].append(
    {"Name": "discount_pct", "Type": "double", "Comment": "Applied discount percentage"}
)

# 4. Push the update — Glue saves the old schema as a version
glue.update_table(DatabaseName="prod_analytics", TableInput=table_input)
print("Column 'discount_pct' added. Old schema saved as a version.")

# 5. List schema versions to see history
versions = glue.get_table_versions(
    DatabaseName="prod_analytics", TableName="sales_transactions"
)
for v in versions["TableVersions"]:
    ncols = len(v["Table"]["StorageDescriptor"]["Columns"])
    print(f"  Version {v['VersionId']}: {ncols} columns ({v['Table']['UpdateTime']})")

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("GlueCatalogExample")
    # Tell Spark to use Glue Catalog instead of local Hive Metastore
    .config("spark.sql.catalogImplementation", "hive")
    .config("hive.metastore.client.factory.class",
            "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory")
    .enableHiveSupport()
    .getOrCreate()
)

# Now Spark SQL sees the same tables as Athena
spark.sql("SHOW DATABASES").show()
spark.sql("USE prod_analytics")
spark.sql("SHOW TABLES").show()

# Read directly using catalog table name — no path needed!
df = spark.sql("SELECT * FROM sales_transactions WHERE year='2024'")
df.show(5)

import boto3
from datetime import date, timedelta

glue = boto3.client("glue")
DB   = "prod_analytics"
TBL  = "sales_transactions"
BASE = "s3://my-data-lake/silver/sales"

# Build partition objects for the last 7 days
partitions = []
today = date.today()
for i in range(7):
    d = today - timedelta(days=i)
    yr, mo = str(d.year), f"{d.month:02d}"
    partitions.append({
        "Values": [yr, mo],               # must match PartitionKeys order
        "StorageDescriptor": {
            "Location": f"{BASE}/year={yr}/month={mo}/",
            "InputFormat":  "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
            "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
            "SerdeInfo": {"SerializationLibrary":
                "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"},
            "Compressed": True
        }
    })

# batch_create_partition — up to 100 per call
for chunk in [partitions[i:i+100] for i in range(0, len(partitions), 100)]:
    resp = glue.batch_create_partition(
        DatabaseName=DB, TableName=TBL,
        PartitionInputList=chunk
    )
    if resp.get("Errors"):
        for err in resp["Errors"]:
            # AlreadyExistsException is fine — partition already registered
            if err["ErrorDetail"]["ErrorCode"] != "AlreadyExistsException":
                print(f"ERROR: {err}")

print(f"Registered {len(partitions)} partitions in Glue Catalog.")

import boto3, time

glue = boto3.client("glue")

# ── Create the crawler ───────────────────────────────────────────
glue.create_crawler(
    Name="silver-sales-crawler",
    Role="arn:aws:iam::123456789012:role/glue-crawler-role",
    DatabaseName="prod_analytics",         # target Catalog database
    Targets={
        "S3Targets": [{
            "Path": "s3://my-data-lake/silver/sales/",
            "Exclusions": ["**/_temporary/**", "**/_spark_metadata/**"]
        }]
    },
    TablePrefix="silver_",                  # tables created as silver_sales, etc.
    SchemaChangePolicy={
        "UpdateBehavior": "UPDATE_IN_DATABASE",  # auto-update schema
        "DeleteBehavior": "LOG"                 # log deleted tables, don't auto-delete
    },
    RecrawlPolicy={"RecrawlBehavior": "CRAWL_NEW_FOLDERS_ONLY"},  # incremental
    Schedule="cron(0 6 * * ? *)"            # daily at 06:00 UTC
)

# ── Start the crawler on-demand (outside of schedule) ───────────
glue.start_crawler(Name="silver-sales-crawler")

# ── Poll until READY (no built-in waiter — must poll manually) ──
def wait_for_crawler(name: str, poll_sec: int = 15, timeout: int = 600):
    start = time.time()
    while time.time() - start < timeout:
        state = glue.get_crawler(Name=name)["Crawler"]["State"]
        print(f"  Crawler state: {state}")
        if state == "READY":
            print("✅ Crawler finished.")
            return
        if state == "STOPPING":
            time.sleep(poll_sec)
            continue
        time.sleep(poll_sec)
    raise TimeoutError(f"Crawler did not finish within {timeout}s")

wait_for_crawler("silver-sales-crawler")

import boto3

glue = boto3.client("glue")

# Custom CSV classifier — pipe-delimited, quoted, with header
glue.create_classifier(
    CsvClassifier={
        "Name": "pipe-delimited-csv",
        "Delimiter": "|",
        "QuoteSymbol": '"',
        "ContainsHeader": "PRESENT",    # first row is header
        "AllowSingleColumn": False
    }
)

# Attach to a crawler
glue.create_crawler(
    Name="legacy-csv-crawler",
    Role="arn:aws:iam::123456789012:role/glue-crawler-role",
    DatabaseName="raw_landing",
    Targets={"S3Targets": [{"Path": "s3://my-lake/raw/legacy-exports/"}]},
    Classifiers=["pipe-delimited-csv"]   # applies before built-in classifiers
)

import sys
import boto3
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext
from pyspark.sql import functions as F

# ── 1. Parse job arguments ───────────────────────────────────────
args = getResolvedOptions(sys.argv, [
    "JOB_NAME",        # always required for job bookmarks
    "source_bucket",
    "target_bucket",
    "env"
])

# ── 2. Initialize Glue and Spark contexts ────────────────────────
sc    = SparkContext()
glue  = GlueContext(sc)
spark = glue.spark_session
job   = Job(glue)
job.init(args["JOB_NAME"], args)          # init required for bookmarks

# ── 3. Read data (standard Spark or DynamicFrame) ────────────────
df = spark.read.parquet(
    f"s3://{args['source_bucket']}/bronze/orders/"
)

# ── 4. Transform ─────────────────────────────────────────────────
df_clean = (df
    .dropDuplicates(["order_id"])
    .filter(F.col("amount") > 0)
    .withColumn("load_date", F.current_date())
)

# ── 5. Write ─────────────────────────────────────────────────────
df_clean.write.mode("overwrite").parquet(
    f"s3://{args['target_bucket']}/silver/orders/"
)

# ── 6. Commit job (required — marks bookmark checkpoint) ─────────
job.commit()
print("Job committed successfully.")

import boto3

glue = boto3.client("glue")

response = glue.start_job_run(
    JobName="silver-orders-etl",
    Arguments={
        # User-defined parameters — read by getResolvedOptions()
        "--source_bucket": "my-lake-raw",
        "--target_bucket": "my-lake-silver",
        "--env":           "prod",

        # Spark configuration overrides for this run
        "--conf": (
            "spark.sql.shuffle.partitions=200 "
            "spark.sql.sources.partitionOverwriteMode=dynamic "
            "spark.serializer=org.apache.spark.serializer.KryoSerializer"
        ),

        # Glue-specific options
        "--enable-metrics":                "",   # push metrics to CloudWatch
        "--enable-continuous-cloudwatch-log": "",  # real-time log streaming
        "--job-bookmark-option":           "job-bookmark-enable"
    },
    MaxCapacity=10.0    # DPUs to allocate (10 DPU = 40 vCPUs + 160 GB RAM)
)

run_id = response["JobRunId"]
print(f"Started job run: {run_id}")

import sys
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext

args  = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc    = SparkContext()
glue  = GlueContext(sc)
job   = Job(glue)
job.init(args["JOB_NAME"], args)       # ← bookmark state restored here

# Read using DynamicFrame — bookmark tracking works automatically
# Glue will only read files it hasn't seen in previous successful runs
raw_dyf = glue.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={
        "paths": ["s3://my-lake/raw/landing/"],
        "recurse": True
    },
    format="parquet",
    transformation_ctx="raw_data"      # ← name used as bookmark key
)

print(f"New records this run: {raw_dyf.count()}")

# ... transform and write ...

job.commit()   # ← bookmark state saved — next run picks up from here

# After writing partitioned data in the Glue job, repair the table
# (this scans S3 and adds any missing partitions to the Glue Catalog)
spark.sql("MSCK REPAIR TABLE prod_analytics.sales_transactions")
print("Partitions registered in Glue Catalog.")

# OR — more efficient for daily jobs: only add today's partition
from datetime import date
today = date.today()
spark.sql(f"""
    ALTER TABLE prod_analytics.sales_transactions
    ADD IF NOT EXISTS PARTITION (year='{today.year}', month='{today.month:02d}')
    LOCATION 's3://my-lake/silver/sales/year={today.year}/month={today.month:02d}/'
""")

from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

# Read with mergeSchema — handles files with different column sets
df = spark.read.option("mergeSchema", True).parquet(
    "s3://my-lake/raw/sales/"
)

# Safely handle columns that may not exist in older files
if "discount_pct" not in df.columns:
    df = df.withColumn("discount_pct", F.lit(0.0).cast(DoubleType()))

# Cast to expected types in case source widened (e.g. int → bigint)
df = df.withColumn("order_id", F.col("order_id").cast("bigint"))

print(f"Schema after evolution handling: {df.schema.simpleString()}")

import boto3

glue = boto3.client("glue")

glue.create_connection(
    ConnectionInput={
        "Name": "prod-rds-postgres",
        "Description": "Production RDS PostgreSQL — analytics database",
        "ConnectionType": "JDBC",
        "ConnectionProperties": {
            "JDBC_CONNECTION_URL": "jdbc:postgresql://prod-rds.cluster-abc.us-east-1.rds.amazonaws.com:5432/analytics",
            "USERNAME": "glue_user",
            "PASSWORD": "{{resolve:secretsmanager:prod/rds/glue-user}}"
            # In practice: Glue reads creds from Secrets Manager directly via secret name
        },
        "PhysicalConnectionRequirements": {
            "SubnetId":               "subnet-0abc123",        # private subnet in your VPC
            "SecurityGroupIdList":     ["sg-0xyz789"],           # SG allowing outbound to RDS
            "AvailabilityZone":        "us-east-1a"
        }
    }
)
print("Glue Connection 'prod-rds-postgres' created.")

# Inside the Glue ETL job script — read from RDS via JDBC connection
customers_dyf = glue.create_dynamic_frame.from_catalog(
    database="prod_analytics",
    table_name="raw_customers",                  # Glue Catalog table pointing to RDS
    additional_options={"jobBookmarkKeys": ["updated_at"]}
)

# OR — direct JDBC read with partitioning for parallelism
customers_df = (spark.read
    .format("jdbc")
    .option("url",             "jdbc:postgresql://prod-rds:5432/analytics")
    .option("dbtable",         "public.customers")
    .option("user",            db_user)
    .option("password",        db_pass)
    .option("partitionColumn", "customer_id")   # enables parallel reads
    .option("lowerBound",      "1")
    .option("upperBound",      "10000000")
    .option("numPartitions",   "10")            # 10 parallel JDBC tasks
    .option("driver",          "org.postgresql.Driver")
    .load()
)

# DQDL — Data Quality Definition Language
# This is a string you pass to Glue, not Python

Rules = [
    # Completeness — no nulls in critical columns
    IsComplete "customer_id",
    IsComplete "txn_id",
    IsComplete "amount",

    # Uniqueness — no duplicate transaction IDs
    IsUnique "txn_id",

    # Value ranges — business rules
    ColumnValues "amount" between 0.01 and 1000000,
    ColumnValues "discount_pct" between 0 and 1,

    # Regex matching — basic format checks
    MatchesRegex "email" with pattern "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z]{2,}$",

    # Referential integrity — values must be in an allowed set
    ColumnValues "status" in ["PENDING","COMPLETED","CANCELLED","REFUNDED"],

    # Row count — detect data drops
    RowCount >= 1000,

    # Completeness ratio — allow up to 2% nulls in optional fields
    Completeness "notes" >= 0.98,

    # Data freshness — data should be from last 48 hours
    DataFreshness "created_at" <= 48 hours
]

import boto3, time, json

glue    = boto3.client("glue")
dynamo  = boto3.resource("dynamodb")
tbl     = dynamo.Table("pipeline-dq-audit")

DQDL_RULES = """
Rules = [
    IsComplete "customer_id",
    IsComplete "txn_id",
    IsUnique "txn_id",
    ColumnValues "amount" between 0.01 and 1000000,
    RowCount >= 1000
]
"""

# ── 1. Create the ruleset (idempotent) ──────────────────────────
try:
    glue.create_data_quality_ruleset(
        Name="sales-transactions-dq",
        Ruleset=DQDL_RULES,
        TargetTable={
            "TableName":     "sales_transactions",
            "DatabaseName":  "prod_analytics"
        }
    )
except glue.exceptions.AlreadyExistsException:
    glue.update_data_quality_ruleset(
        Name="sales-transactions-dq", Ruleset=DQDL_RULES
    )

# ── 2. Run the evaluation ────────────────────────────────────────
run_resp = glue.start_data_quality_ruleset_evaluation_run(
    DataSource={"GlueTable": {
        "TableName": "sales_transactions", "DatabaseName": "prod_analytics"
    }},
    Role="arn:aws:iam::123456789012:role/glue-dq-role",
    RulesetNames=["sales-transactions-dq"],
    NumberOfWorkers=5
)
run_id = run_resp["RunId"]

# ── 3. Poll until finished ───────────────────────────────────────
for _ in range(40):
    run_detail = glue.get_data_quality_ruleset_evaluation_run(RunId=run_id)
    status = run_detail["Status"]
    print(f"DQ run status: {status}")
    if status in ["SUCCEEDED", "FAILED", "ERROR"]:
        break
    time.sleep(15)

# ── 4. Check rule results ────────────────────────────────────────
results = run_detail.get("ResultIds", [])
overall_pass = True
dq_score     = 0.0

if results:
    result_detail = glue.get_data_quality_result(ResultId=results[0])
    rule_results  = result_detail["RuleResults"]
    score         = result_detail.get("Score", 0.0)
    dq_score      = score

    print(f"\nOverall DQ Score: {score:.2%}")
    for r in rule_results:
        icon = "✅" if r["Result"] == "PASS" else "❌"
        print(f"  {icon} {r['Name']}: {r['Result']} — {r.get('Description','')}")
        if r["Result"] == "FAIL":
            overall_pass = False

# ── 5. Write DQ results to DynamoDB audit table ──────────────────
from datetime import datetime
from decimal import Decimal

tbl.put_item(Item={
    "run_id":   run_id,
    "table":    "sales_transactions",
    "ts":       datetime.utcnow().isoformat(),
    "score":    Decimal(str(round(dq_score, 4))),
    "passed":   overall_pass
})

# ── 6. Fail the pipeline if score below threshold ────────────────
DQ_THRESHOLD = 0.95
if dq_score < DQ_THRESHOLD:
    raise ValueError(
        f"DQ score {dq_score:.2%} below threshold {DQ_THRESHOLD:.0%}. Pipeline halted."
    )

import boto3, time
from botocore.exceptions import ClientError

glue = boto3.client("glue")

# ── CREATE a Glue Spark job ───────────────────────────────────────
glue.create_job(
    Name="silver-orders-etl",
    Role="arn:aws:iam::123456789012:role/glue-etl-role",
    Command={
        "Name":          "glueetl",               # Spark job type
        "ScriptLocation": "s3://my-scripts/etl/silver_orders.py",
        "PythonVersion":  "3"
    },
    GlueVersion="4.0",                         # Spark 3.3 + Python 3.10
    MaxCapacity=10.0,                           # DPUs
    Timeout=60,                                 # minutes before auto-kill
    DefaultArguments={
        "--job-bookmark-option": "job-bookmark-enable",
        "--enable-metrics":      "",
        "--env":                 "prod"
    },
    Connections={"Connections": ["prod-rds-postgres"]}
)

# ── START a job run ───────────────────────────────────────────────
resp   = glue.start_job_run(JobName="silver-orders-etl")
run_id = resp["JobRunId"]
print(f"Started: {run_id}")

# ── POLL until terminal state ─────────────────────────────────────
TERMINAL = {"SUCCEEDED", "FAILED", "ERROR", "TIMEOUT", "STOPPED"}
while True:
    run = glue.get_job_run(JobName="silver-orders-etl", RunId=run_id)
    state = run["JobRun"]["JobRunState"]
    print(f"  State: {state}")
    if state in TERMINAL:
        break
    time.sleep(30)

if state != "SUCCEEDED":
    error = run["JobRun"].get("ErrorMessage", "unknown")
    raise RuntimeError(f"Glue job {run_id} {state}: {error}")

# ── LIST run history with paginator ──────────────────────────────
paginator = glue.get_paginator("get_job_runs")
for page in paginator.paginate(JobName="silver-orders-etl"):
    for run in page["JobRuns"]:
        print(f"  {run['JobRunId']}: {run['JobRunState']} — {run.get('CompletedOn','')}")

# ── STOP a running job ────────────────────────────────────────────
glue.batch_stop_job_run(
    JobName="silver-orders-etl",
    JobRunIds=[run_id]
)

# Submit a PySpark job to a running EMR cluster
aws emr add-steps \
  --cluster-id j-2AXXXXXXGAPLF \
  --steps '[{
    "Name": "Daily Sales ETL",
    "ActionOnFailure": "CONTINUE",
    "HadoopJarStep": {
      "Jar": "command-runner.jar",
      "Args": [
        "spark-submit",
        "--deploy-mode", "cluster",
        "--conf", "spark.executor.memory=8g",
        "--conf", "spark.executor.cores=4",
        "--conf", "spark.sql.shuffle.partitions=200",
        "s3://my-code-bucket/scripts/daily_sales_etl.py",
        "--date", "2024-01-15",
        "--env",  "prod"
      ]
    }
  }]'

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SalesETL").getOrCreate()

# On EMR, s3:// paths work exactly like local paths
# EMRFS handles the translation transparently
df = spark.read.parquet("s3://my-data-lake/bronze/sales/year=2024/")

result = df.groupBy("region").agg({"revenue": "sum"})

# Write results back to S3 — persists after cluster terminates
result.write.mode("overwrite").parquet(
    "s3://my-data-lake/gold/sales_by_region/year=2024/"
)

spark.stop()

[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.sql.shuffle.partitions":        "400",
      "spark.default.parallelism":           "400",
      "spark.executor.memory":              "8g",
      "spark.executor.cores":               "4",
      "spark.driver.memory":                "4g",
      "spark.dynamicAllocation.enabled":    "true",
      "spark.sql.adaptive.enabled":         "true",
      "spark.serializer":                   "org.apache.spark.serializer.KryoSerializer"
    }
  },
  {
    "Classification": "spark-env",
    "Configurations": [{
      "Classification": "export",
      "Properties": {
        "PYSPARK_PYTHON": "/usr/bin/python3"
      }
    }]
  }
]

import boto3, time

emr = boto3.client("emr-serverless", region_name="us-east-1")

# ── Step 1: Create an Application (one-time setup) ──────────────
app = emr.create_application(
    name="spark-etl-app",
    releaseLabel="emr-6.15.0",         # EMR runtime version
    type="SPARK",
    autoStartConfiguration={"enabled": True},
    autoStopConfiguration={
        "enabled": True,
        "idleTimeoutMinutes": 15          # auto-stop if idle 15 min
    },
    maximumCapacity={                        # cost guard-rail
        "cpu":    "200 vCPU",
        "memory": "1000 GB"
    }
)
app_id = app["applicationId"]
print(f"Application created: {app_id}")

# ── Step 2: Start the Application ───────────────────────────────
emr.start_application(applicationId=app_id)
# Wait until it's in STARTED state
while True:
    state = emr.get_application(applicationId=app_id)["application"]["state"]
    if state == "STARTED": break
    time.sleep(5)

# ── Step 3: Submit a Job Run ─────────────────────────────────────
job = emr.start_job_run(
    applicationId=app_id,
    executionRoleArn="arn:aws:iam::123456789012:role/EMRServerlessJobRole",
    jobDriver={
        "sparkSubmit": {
            "entryPoint": "s3://my-code-bucket/scripts/daily_etl.py",
            "entryPointArguments": ["--date", "2024-01-15"],
            "sparkSubmitParameters": (
                "--conf spark.executor.cores=4 "
                "--conf spark.executor.memory=8g "
                "--conf spark.sql.shuffle.partitions=200"
            )
        }
    },
    configurationOverrides={
        "monitoringConfiguration": {
            "s3MonitoringConfiguration": {
                "logUri": "s3://my-emr-logs/serverless/"
            }
        }
    },
    name="daily-etl-2024-01-15"
)
job_run_id = job["jobRunId"]
print(f"Job submitted: {job_run_id}")

# ── Step 4: Poll until complete ──────────────────────────────────
while True:
    run = emr.get_job_run(applicationId=app_id, jobRunId=job_run_id)
    state = run["jobRun"]["state"]
    print(f"State: {state}")
    if state in ("SUCCESS", "FAILED", "CANCELLED"): break
    time.sleep(20)

print(f"Final state: {state}")

import boto3

emr = boto3.client("emr")

emr.put_managed_scaling_policy(
    ClusterId="j-2AXXXXXXGAPLF",
    ManagedScalingPolicy={
        "ComputeLimits": {
            "UnitType":            "Instances",
            "MinimumCapacityUnits": 2,    # minimum 2 core/task nodes
            "MaximumCapacityUnits": 20,   # scale up to 20 nodes
            "MaximumOnDemandCapacityUnits": 5, # max on-demand (rest Spot)
            "MaximumCoreCapacityUnits": 5  # core nodes stay small
        }
    }
)

import boto3

emr = boto3.client("emr")

response = emr.run_job_flow(
    Name="production-spark-cluster",
    ReleaseLabel="emr-7.1.0",
    Applications=[{"Name": "Spark"}],
    Instances={
        "InstanceFleets": [
            {
                "Name": "MasterFleet",
                "InstanceFleetType": "MASTER",
                "TargetOnDemandCapacity": 1,  # Master always on-demand
                "InstanceTypeConfigs": [
                    {"InstanceType": "m5.xlarge"}
                ]
            },
            {
                "Name": "CoreFleet",
                "InstanceFleetType": "CORE",
                "TargetOnDemandCapacity": 2,   # Core on-demand
                "InstanceTypeConfigs": [
                    {"InstanceType": "r5.4xlarge", "WeightedCapacity": 1},
                    {"InstanceType": "r5a.4xlarge", "WeightedCapacity": 1}
                ]
            },
            {
                "Name": "TaskFleet",
                "InstanceFleetType": "TASK",
                "TargetSpotCapacity": 10,  # Task nodes on Spot!
                "InstanceTypeConfigs": [
                    # Multiple types = better Spot availability
                    {"InstanceType": "r5.4xlarge",  "WeightedCapacity": 1},
                    {"InstanceType": "r5a.4xlarge", "WeightedCapacity": 1},
                    {"InstanceType": "r4.4xlarge",  "WeightedCapacity": 1},
                    {"InstanceType": "m5.8xlarge",  "WeightedCapacity": 2}
                ],
                "LaunchSpecifications": {
                    "SpotSpecification": {
                        "TimeoutDurationMinutes": 5,
                        "TimeoutAction": "SWITCH_TO_ON_DEMAND"  # fallback
                    }
                }
            }
        ],
        "Ec2SubnetIds": ["subnet-abc123", "subnet-def456"],  # multi-AZ
        "KeepJobFlowAliveWhenNoSteps": False,  # auto-terminate after steps
    },
    JobFlowRole="EMR_EC2_DefaultRole",
    ServiceRole="EMR_DefaultRole",
    AutoTerminationPolicy={"IdleTimeout": 3600}  # terminate if idle 1 hr
)

cluster_id = response["JobFlowId"]
print(f"Cluster started: {cluster_id}")

#!/bin/bash
# Bootstrap script — runs on every node before EMR services start
set -ex  # exit on error, print each command

# Update pip
sudo pip3 install --upgrade pip

# Install Python dependencies your PySpark job needs
sudo pip3 install \
    pyarrow==14.0.1 \
    great-expectations==0.18.0 \
    boto3==1.34.0 \
    tenacity==8.2.3 \
    psycopg2-binary==2.9.9

# Download a shared config file from S3
aws s3 cp s3://my-config-bucket/pipeline.yaml /etc/pipeline/pipeline.yaml

# Set environment variable for all processes
echo 'export PIPELINE_ENV=prod' | sudo tee -a /etc/environment

echo "Bootstrap complete!"

import boto3

# First, upload your bootstrap script to S3
s3 = boto3.client("s3")
s3.upload_file("bootstrap.sh", "my-code-bucket", "bootstrap/bootstrap.sh")

emr = boto3.client("emr")

response = emr.run_job_flow(
    Name="cluster-with-bootstrap",
    ReleaseLabel="emr-7.1.0",
    Applications=[{"Name": "Spark"}],
    BootstrapActions=[
        {
            "Name": "Install Python Dependencies",
            "ScriptBootstrapAction": {
                "Path": "s3://my-code-bucket/bootstrap/bootstrap.sh",
                "Args": []  # optional args passed to the script
            }
        }
    ],
    Instances={
        "InstanceGroups": [
            {"Name": "Master", "InstanceRole": "MASTER",
             "InstanceType": "m5.xlarge", "InstanceCount": 1},
            {"Name": "Core",   "InstanceRole": "CORE",
             "InstanceType": "r5.4xlarge", "InstanceCount": 4}
        ],
        "Ec2SubnetId": "subnet-abc123",
        "KeepJobFlowAliveWhenNoSteps": False
    },
    JobFlowRole="EMR_EC2_DefaultRole",
    ServiceRole="EMR_DefaultRole",
    LogUri="s3://my-emr-logs/"
)

print(f"Cluster: {response['JobFlowId']}")

import boto3, time

emr = boto3.client("emr", region_name="us-east-1")

# ── Launch cluster with embedded steps ──────────────────────────
response = emr.run_job_flow(
    Name="daily-etl-2024-01-15",
    ReleaseLabel="emr-7.1.0",
    Applications=[{"Name": "Spark"}],
    Instances={
        "InstanceGroups": [
            {
                "Name": "Master", "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge", "InstanceCount": 1,
                "Market": "ON_DEMAND"
            },
            {
                "Name": "Core", "InstanceRole": "CORE",
                "InstanceType": "r5.4xlarge", "InstanceCount": 4,
                "Market": "ON_DEMAND"
            },
            {
                "Name": "Task", "InstanceRole": "TASK",
                "InstanceType": "r5.4xlarge", "InstanceCount": 8,
                "Market": "SPOT",
                "BidPrice": "0.50"  # max Spot bid
            }
        ],
        "Ec2SubnetId": "subnet-abc123",
        "KeepJobFlowAliveWhenNoSteps": False,  # ← auto-terminate!
        "TerminationProtected": False
    },
    Steps=[
        {
            "Name": "Bronze ETL",
            "ActionOnFailure": "TERMINATE_CLUSTER",  # fail fast
            "HadoopJarStep": {
                "Jar": "command-runner.jar",
                "Args": [
                    "spark-submit", "--deploy-mode", "cluster",
                    "--conf", "spark.sql.shuffle.partitions=400",
                    "s3://my-code/scripts/bronze_etl.py",
                    "--date", "2024-01-15"
                ]
            }
        },
        {
            "Name": "Silver Transform",
            "ActionOnFailure": "TERMINATE_CLUSTER",
            "HadoopJarStep": {
                "Jar": "command-runner.jar",
                "Args": [
                    "spark-submit", "--deploy-mode", "cluster",
                    "s3://my-code/scripts/silver_transform.py",
                    "--date", "2024-01-15"
                ]
            }
        }
    ],
    BootstrapActions=[{
        "Name": "Install libs",
        "ScriptBootstrapAction": {
            "Path": "s3://my-code/bootstrap/setup.sh"
        }
    }],
    Configurations=[{
        "Classification": "spark-defaults",
        "Properties": {
            "spark.sql.adaptive.enabled": "true",
            "spark.dynamicAllocation.enabled": "true"
        }
    }],
    JobFlowRole="EMR_EC2_DefaultRole",
    ServiceRole="EMR_DefaultRole",
    LogUri="s3://my-emr-logs/clusters/",
    AutoTerminationPolicy={"IdleTimeout": 3600},
    Tags=[
        {"Key": "Project",     "Value": "DataLake"},
        {"Key": "Environment", "Value": "prod"},
        {"Key": "CostCenter",   "Value": "DE-Team"}  # for billing reports
    ]
)

cluster_id = response["JobFlowId"]
print(f"Cluster launched: {cluster_id}")
print(f"Will auto-terminate after steps complete.")

# ── Poll until cluster terminates ───────────────────────────────
while True:
    cluster = emr.describe_cluster(ClusterId=cluster_id)
    state   = cluster["Cluster"]["Status"]["State"]
    print(f"  Cluster state: {state}")
    if state in ("TERMINATED", "TERMINATED_WITH_ERRORS"):
        break
    time.sleep(30)

print("Done.")

-- Run this DDL in Athena console or via start_query_execution API
CREATE EXTERNAL TABLE IF NOT EXISTS sales_db.transactions (
    transaction_id  STRING,
    customer_id     STRING,
    product_id      STRING,
    amount          DOUBLE,
    status          STRING,
    created_at      TIMESTAMP
)
PARTITIONED BY (
    year  STRING,
    month STRING,
    day   STRING
)
STORED AS PARQUET
LOCATION 's3://my-data-lake/silver/transactions/'
TBLPROPERTIES (
    'parquet.compress' = 'SNAPPY',
    'projection.enabled' = 'false'
);

-- Load partition metadata (tells Athena about existing partitions)
MSCK REPAIR TABLE sales_db.transactions;

-- Now query with partition pruning (only scans year=2024/month=01/day=15)
SELECT
    customer_id,
    SUM(amount) AS total_spend
FROM sales_db.transactions
WHERE
    year  = '2024'
    AND month = '01'
    AND day   = '15'
    AND status = 'COMPLETED'
GROUP BY customer_id
ORDER BY total_spend DESC
LIMIT 100;

-- ❌ BAD: Full table scan — reads ALL partitions regardless of date
SELECT * FROM sales_db.transactions
WHERE created_at >= TIMESTAMP '2024-01-15 00:00:00';
-- Athena scans ALL files → slow + expensive
-- created_at is a data column, not a partition column

-- ✅ GOOD: Partition pruning — only reads year=2024/month=01/day=15
SELECT * FROM sales_db.transactions
WHERE year = '2024' AND month = '01' AND day = '15'
  AND created_at >= TIMESTAMP '2024-01-15 00:00:00';
-- Athena skips all other partitions → fast + cheap

-- ✅ ALSO GOOD: Range on partition columns
SELECT * FROM sales_db.transactions
WHERE year = '2024' AND month IN ('01', '02', '03');
-- Scans only Q1 2024 data

CREATE EXTERNAL TABLE sales_db.events (
    event_id   STRING,
    user_id    STRING,
    event_type STRING,
    payload    STRING
)
PARTITIONED BY (dt STRING)   -- single date partition column
STORED AS PARQUET
LOCATION 's3://my-data-lake/events/'
TBLPROPERTIES (
    -- Enable partition projection
    'projection.enabled'               = 'true',
    'projection.dt.type'               = 'date',
    'projection.dt.format'             = 'yyyy-MM-dd',
    'projection.dt.range'              = '2023-01-01,NOW',
    'projection.dt.interval'           = '1',
    'projection.dt.interval.unit'      = 'DAYS',

    -- Tell Athena how to build the S3 path from partition value
    'storage.location.template'        = 's3://my-data-lake/events/dt=${dt}/'
);

-- No MSCK REPAIR needed! Just query directly:
SELECT event_type, COUNT(*) AS cnt
FROM  sales_db.events
WHERE dt BETWEEN '2024-01-01' AND '2024-01-31'
GROUP BY 1
ORDER BY 2 DESC;

import boto3

athena = boto3.client("athena", region_name="us-east-1")

athena.create_work_group(
    Name="de-team-prod",
    Description="Data Engineering team production workgroup",
    Configuration={
        "ResultConfiguration": {
            "OutputLocation": "s3://my-athena-results/de-team-prod/",
            "EncryptionConfiguration": {
                "EncryptionOption": "SSE_KMS",
                "KmsKey": "alias/data-lake-cmk"
            }
        },
        "EnforceWorkGroupConfiguration": True,  # users can't override below
        "BytesScannedCutoffPerQuery": 10 * 1024**3,  # 10 GB max per query
        "PublishCloudWatchMetricsEnabled": True,  # send metrics to CW
        "RequesterPaysEnabled": False
    },
    Tags=[{"Key": "Team", "Value": "DataEngineering"}]
)
print("Workgroup created.")

import boto3

athena = boto3.client("athena")

athena.update_work_group(
    WorkGroup="de-team-prod",
    ConfigurationUpdates={
        "ResultReuseByAgeConfiguration": {
            "Enabled":           True,
            "MaxAgeInMinutes":   60 * 24  # cache results for 24 hours
        }
    }
)

-- Create an Iceberg table (stored on S3, managed via Glue Catalog)
CREATE TABLE sales_db.customers_iceberg (
    customer_id  STRING,
    name         STRING,
    email        STRING,
    country      STRING,
    updated_at   TIMESTAMP
)
LOCATION 's3://my-data-lake/iceberg/customers/'
TBLPROPERTIES (
    'table_type' = 'ICEBERG',
    'format'     = 'parquet',
    'write_compression' = 'snappy'
);

-- Standard INSERT
INSERT INTO sales_db.customers_iceberg
VALUES ('C001', 'Alice', 'alice@example.com', 'IN', NOW());

-- MERGE INTO — upsert from a staging table
MERGE INTO sales_db.customers_iceberg t
USING      sales_db.customers_staging s
ON         t.customer_id = s.customer_id
WHEN MATCHED THEN
    UPDATE SET name = s.name, email = s.email, updated_at = s.updated_at
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, country, updated_at)
    VALUES (s.customer_id, s.name, s.email, s.country, s.updated_at);

-- DELETE rows (e.g. GDPR right-to-erasure)
DELETE FROM sales_db.customers_iceberg
WHERE customer_id = 'C001';

-- Time travel — query historical snapshot
SELECT * FROM sales_db.customers_iceberg
FOR TIMESTAMP AS OF TIMESTAMP '2024-01-15 12:00:00';

-- After installing the RDS connector Lambda and registering it as a catalog:
-- "my_rds_catalog" points to an RDS PostgreSQL instance
-- "AwsDataCatalog" points to Glue Catalog (S3 data)

SELECT
    t.transaction_id,
    t.amount,
    c.name        AS customer_name,
    c.email       AS customer_email,
    c.country
FROM
    -- S3 data via Glue Catalog
    AwsDataCatalog.sales_db.transactions t

    -- RDS PostgreSQL data via federated connector
    JOIN my_rds_catalog.public.customers c
      ON t.customer_id = c.customer_id
WHERE
    t.year   = '2024'
    AND t.month  = '01'
    AND c.country = 'IN';

import boto3

athena = boto3.client("athena")

# Save a frequently-used DQ check as a named query
response = athena.create_named_query(
    Name="daily_null_check",
    Description="Check null counts in transactions table for a given date",
    Database="sales_db",
    QueryString="""
        SELECT
            COUNT(*)                                      AS total_rows,
            COUNT(*) - COUNT(transaction_id)              AS null_transaction_id,
            COUNT(*) - COUNT(customer_id)                 AS null_customer_id,
            COUNT(*) - COUNT(amount)                      AS null_amount
        FROM transactions
        WHERE year = '2024' AND month = '01' AND day = '15'
    """,
    WorkGroup="de-team-prod"
)
named_query_id = response["NamedQueryId"]
print(f"Saved named query: {named_query_id}")

# Retrieve it later to get the SQL string
nq = athena.get_named_query(NamedQueryId=named_query_id)
sql = nq["NamedQuery"]["QueryString"]
print(f"SQL: {sql}")

import boto3, time

athena = boto3.client("athena", region_name="us-east-1")

# ─────────────────────────────────────────────────────────────────
# Step 1: Submit the query
# ─────────────────────────────────────────────────────────────────
response = athena.start_query_execution(
    QueryString="""
        SELECT
            region,
            SUM(revenue)   AS total_revenue,
            COUNT(*)       AS num_orders
        FROM sales_db.transactions
        WHERE year = '2024' AND month = '01'
        GROUP BY region
        ORDER BY total_revenue DESC
    """,
    QueryExecutionContext={
        "Database": "sales_db",
        "Catalog":  "AwsDataCatalog"
    },
    ResultConfiguration={
        "OutputLocation": "s3://my-athena-results/de-team-prod/"
    },
    WorkGroup="de-team-prod"
)

query_execution_id = response["QueryExecutionId"]
print(f"Query submitted: {query_execution_id}")

# ─────────────────────────────────────────────────────────────────
# Step 2: Poll until SUCCEEDED or FAILED
# ─────────────────────────────────────────────────────────────────
def wait_for_query(athena_client, qeid, poll_interval=2):
    terminal = {"SUCCEEDED", "FAILED", "CANCELLED"}
    while True:
        result = athena_client.get_query_execution(QueryExecutionId=qeid)
        status = result["QueryExecution"]["Status"]
        state  = status["State"]
        print(f"  State: {state}")
        if state == "FAILED":
            reason = status.get("StateChangeReason", "Unknown")
            raise RuntimeError(f"Athena query FAILED: {reason}")
        if state == "CANCELLED":
            raise RuntimeError("Athena query was CANCELLED.")
        if state == "SUCCEEDED":
            stats = result["QueryExecution"]["Statistics"]
            scanned_mb = stats["DataScannedInBytes"] / 1024**2
            print(f"  ✅ SUCCEEDED — Scanned: {scanned_mb:.1f} MB")
            return
        time.sleep(poll_interval)

wait_for_query(athena, query_execution_id)

# ─────────────────────────────────────────────────────────────────
# Step 3: Fetch and parse results with paginator
# ─────────────────────────────────────────────────────────────────
def fetch_results(athena_client, qeid):
    paginator = athena_client.get_paginator("get_query_results")
    pages = paginator.paginate(QueryExecutionId=qeid)

    rows = []
    headers = None

    for page in pages:
        result_set = page["ResultSet"]
        if headers is None:
            # First row in first page is the header row
            headers = [
                col["VarCharValue"]
                for col in result_set["Rows"][0]["Data"]
            ]
            data_rows = result_set["Rows"][1:]  # skip header
        else:
            data_rows = result_set["Rows"]

        for row in data_rows:
            values = [
                cell.get("VarCharValue", None)
                for cell in row["Data"]
            ]
            rows.append(dict(zip(headers, values)))

    return rows

results = fetch_results(athena, query_execution_id)

# Step 4: Use results — print or convert to DataFrame
for row in results:
    print(row)
# → {'region': 'South', 'total_revenue': '1234567.89', 'num_orders': '4521'}
# → {'region': 'North', 'total_revenue': '987654.32', 'num_orders': '3200'}

# Convert to Pandas DataFrame if needed
import pandas as pd
df = pd.DataFrame(results)
df["total_revenue"] = df["total_revenue"].astype(float)
print(df)

import boto3

athena = boto3.client("athena")

# Cancel a running query (e.g. if it's scanning too much data)
athena.stop_query_execution(
    QueryExecutionId="abc12345-1234-1234-1234-abc123456789"
)
print("Query cancellation requested.")

import boto3

lf = boto3.client("lakeformation", region_name="us-east-1")

# Register the S3 path — Lake Formation now controls access here
lf.register_resource(
    ResourceArn="arn:aws:s3:::my-data-lake",
    UseServiceLinkedRole=True  # Lake Formation uses its own IAM role to access S3
)
print("S3 location registered with Lake Formation.")

# List all registered locations
response = lf.list_resources()
for r in response["ResourceInfoList"]:
    print(r["ResourceArn"], r["RoleArn"])

import boto3

lf = boto3.client("lakeformation")

# Grant SELECT on the transactions table to the Data Analyst role
lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/DataAnalystRole"
    },
    Resource={
        "Table": {
            "CatalogId":  "123456789012",
            "DatabaseName": "sales_db",
            "Name":         "transactions"
        }
    },
    Permissions=["SELECT"],
    PermissionsWithGrantOption=[]   # cannot re-grant to others
)
print("Table permission granted.")

# Grant SELECT on ALL tables in a database
lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/DataEngineerRole"
    },
    Resource={
        "Database": {
            "CatalogId":    "123456789012",
            "Name": "sales_db"
        }
    },
    Permissions=["ALL"]   # full access to the database
)

# Revoke a permission
lf.revoke_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/DataAnalystRole"
    },
    Resource={
        "Table": {
            "CatalogId": "123456789012",
            "DatabaseName": "sales_db",
            "Name": "transactions"
        }
    },
    Permissions=["SELECT"]
)

import boto3

lf = boto3.client("lakeformation")

# customers table has: customer_id, name, email, phone, country, segment
# Analyst role should NOT see email or phone (PII)

lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/DataAnalystRole"
    },
    Resource={
        "TableWithColumns": {
            "CatalogId":    "123456789012",
            "DatabaseName": "sales_db",
            "Name":         "customers",
            "ColumnNames": [
                "customer_id",
                "name",
                "country",
                "segment"
                # email and phone are NOT listed → access denied
            ]
        }
    },
    Permissions=["SELECT"]
)
print("Column-level permission granted (email and phone excluded).")

# Alternatively, use ColumnWildcard with Excluded columns
# (grant all columns EXCEPT the listed ones)
lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/DataAnalystRole"
    },
    Resource={
        "TableWithColumns": {
            "CatalogId":    "123456789012",
            "DatabaseName": "sales_db",
            "Name":         "customers",
            "ColumnWildcard": {
                "ExcludedColumnNames": ["email", "phone"]
            }
        }
    },
    Permissions=["SELECT"]
)

import boto3

lf = boto3.client("lakeformation")

# Step 1: Create a row filter (a named WHERE clause)
lf.create_data_cells_filter(
    TableData={
        "TableCatalogId": "123456789012",
        "DatabaseName":   "sales_db",
        "TableName":      "transactions",
        "Name":           "india_only_filter",  # filter name
        "RowFilter": {
            "FilterExpression": "country = 'IN'"  # SQL WHERE clause
        },
        # Optionally restrict columns too in the same filter
        "ColumnWildcard": {}  # all columns allowed in this filter
    }
)
print("Row filter created.")

# Step 2: Grant the filter to an IAM role
# The India Analytics team can only see rows where country = 'IN'
lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/IndiaAnalyticsRole"
    },
    Resource={
        "DataCellsFilter": {
            "TableCatalogId": "123456789012",
            "DatabaseName":   "sales_db",
            "TableName":      "transactions",
            "Name":           "india_only_filter"
        }
    },
    Permissions=["SELECT"]
)
print("Row-level filter grant applied.")
# Now when IndiaAnalyticsRole queries transactions in Athena,
# they ONLY see rows where country = 'IN' — automatically enforced.

import boto3

lf = boto3.client("lakeformation")

# ── Step 1: Create LF-Tag keys and their allowed values ─────────
lf.create_lf_tag(
    TagKey="sensitivity",
    TagValues=["public", "internal", "confidential", "restricted"]
)

lf.create_lf_tag(
    TagKey="domain",
    TagValues=["sales", "finance", "marketing", "hr"]
)
print("LF-Tags created.")

# ── Step 2: Assign tags to a database ───────────────────────────
lf.add_lf_tags_to_resource(
    Resource={
        "Database": {
            "CatalogId": "123456789012",
            "Name": "sales_db"
        }
    },
    LFTags=[
        {"TagKey": "domain",      "TagValues": ["sales"]},
        {"TagKey": "sensitivity", "TagValues": ["internal"]}
    ]
)

# ── Step 3: Assign a more restrictive tag to a specific table ────
lf.add_lf_tags_to_resource(
    Resource={
        "Table": {
            "CatalogId":    "123456789012",
            "DatabaseName": "sales_db",
            "Name":         "customers"  # has PII
        }
    },
    LFTags=[
        {"TagKey": "sensitivity", "TagValues": ["confidential"]}
    ]
)

# ── Step 4: Grant access via tag expression ──────────────────────
# Sales team can access all tables tagged domain=sales AND sensitivity=internal
lf.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::123456789012:role/SalesAnalyticsRole"
    },
    Resource={
        "LFTagPolicy": {
            "CatalogId":    "123456789012",
            "ResourceType": "TABLE",
            "Expression": [
                {"TagKey": "domain",      "TagValues": ["sales"]},
                {"TagKey": "sensitivity", "TagValues": ["public", "internal"]}
            ]
        }
    },
    Permissions=["SELECT"]
)
print("Tag-based grant applied.")
# SalesAnalyticsRole can now query any table tagged
# domain=sales AND (sensitivity=public OR sensitivity=internal)
# The 'customers' table (tagged confidential) is automatically excluded.
# Any NEW table added to sales_db with these tags is automatically accessible.

import boto3

# Run in ACCOUNT A (data producer)
lf_producer = boto3.client("lakeformation", region_name="us-east-1")

# Grant SELECT on a specific table to Account B's consumer role
lf_producer.grant_permissions(
    Principal={
        "DataLakePrincipalIdentifier":
            "arn:aws:iam::222222222222:role/ConsumerRole"
    },
    Resource={
        "Table": {
            "CatalogId":    "111111111111",
            "DatabaseName": "sales_db",
            "Name":         "transactions"
        }
    },
    Permissions=["SELECT"],
    PermissionsWithGrantOption=["SELECT"]  # allow B to re-share
)
print("Cross-account grant sent to Account B.")

# ────────────────────────────────────────────────────────────────
# In ACCOUNT B: create a resource link to access the shared table
# ────────────────────────────────────────────────────────────────
glue_consumer = boto3.client("glue", region_name="us-east-1")

# Create a local resource link pointing to Account A's table
glue_consumer.create_table(
    DatabaseName="shared_data",
    TableInput={
        "Name": "transactions_from_a",  # local alias
        "TargetTable": {
            "CatalogId":    "111111111111",  # Account A's ID
            "DatabaseName": "sales_db",
            "Name":         "transactions"
        }
    }
)
print("Resource link created. Account B can now query Account A's table.")

import boto3

lf = boto3.client("lakeformation")

# Designate a role as a Data Lake Administrator
# (admins can grant any permission to anyone)
lf.put_data_lake_settings(
    DataLakeSettings={
        "DataLakeAdmins": [
            {"DataLakePrincipalIdentifier":
                "arn:aws:iam::123456789012:role/DataLakeAdminRole"},
            {"DataLakePrincipalIdentifier":
                "arn:aws:iam::123456789012:user/de-lead"}
        ],
        "CreateDatabaseDefaultPermissions": [],  # no implicit grants
        "CreateTableDefaultPermissions":    [],  # no implicit grants
        "TrustedResourceOwners": []
    }
)
print("Data lake admins configured.")

# List all permissions currently in effect
paginator = lf.get_paginator("list_permissions")
for page in paginator.paginate():
    for perm in page["PrincipalResourcePermissions"]:
        principal = perm["Principal"]["DataLakePrincipalIdentifier"]
        resource  = perm["Resource"]
        perms     = perm["Permissions"]
        print(f"  {principal} → {perms} on {resource}")

-- Load all Parquet files under a prefix in parallel
COPY analytics.orders
FROM 's3://my-data-lake/silver/orders/year=2024/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET;

-- Load a specific manifest file (list of exact S3 paths)
COPY analytics.orders
FROM 's3://my-data-lake/manifests/orders_2024_load.manifest'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET
MANIFEST;

COPY analytics.customers
FROM 's3://my-data-lake/raw/customers/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS CSV
DELIMITER ','
QUOTE '"'
IGNOREHEADER 1
GZIP
TIMEFORMAT 'auto'
DATEFORMAT  'auto'
TRUNCATECOLUMNS       -- silently truncate strings that exceed column length
MAXERROR 10;          -- allow up to 10 bad rows before failing

-- Check what was rejected (stl_load_errors is Redshift's error log)
SELECT filename, err_reason, raw_line
FROM   stl_load_errors
ORDER BY starttime DESC
LIMIT  20;

import boto3, time

redshift_data = boto3.client("redshift-data", region_name="us-east-1")

copy_sql = """
COPY analytics.orders
FROM 's3://my-data-lake/silver/orders/year=2024/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET;
"""

# Submit the COPY statement
resp = redshift_data.execute_statement(
    ClusterIdentifier="my-redshift-cluster",
    Database="analytics",
    DbUser="etl_user",
    Sql=copy_sql
)
stmt_id = resp["Id"]

# Poll until FINISHED or FAILED
while True:
    status = redshift_data.describe_statement(Id=stmt_id)["Status"]
    print(f"COPY status: {status}")
    if status in ("FINISHED", "FAILED", "ABORTED"):
        break
    time.sleep(5)

if status == "FINISHED":
    print("✅ COPY completed successfully")
else:
    err = redshift_data.describe_statement(Id=stmt_id).get("Error")
    raise RuntimeError(f"COPY failed: {err}")

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("RedshiftWriter") \
    .config("spark.jars.packages",
            "io.github.spark-redshift-community:spark-redshift_2.12:6.2.0-spark_3.5") \
    .getOrCreate()

df = spark.read.parquet("s3://my-data-lake/silver/orders/")

# Write DataFrame → S3 staging → Redshift COPY (all automatic)
df.write \
  .format("io.github.spark_redshift_community.spark.redshift") \
  .option("url",            "jdbc:redshift://my-cluster.abc.us-east-1.redshift.amazonaws.com:5439/analytics") \
  .option("dbtable",        "analytics.orders") \
  .option("tempdir",        "s3://my-temp-bucket/redshift-staging/") \
  .option("aws_iam_role",   "arn:aws:iam::123456789012:role/RedshiftS3ReadRole") \
  .option("user",           "etl_user") \
  .option("password",       db_password) \
  .mode("append") \           # or "overwrite" to truncate first
  .save()

print("DataFrame written to Redshift via COPY")

-- Export aggregated sales summary to S3 as compressed Parquet
UNLOAD ('
  SELECT  order_date,
          product_category,
          SUM(amount)    AS total_amount,
          COUNT(*)       AS order_count
  FROM    analytics.orders
  WHERE   order_date >= \'\'2024-01-01\'\'
  GROUP BY 1, 2
')
TO 's3://my-data-lake/exports/sales_summary/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET
ALLOWOVERWRITE    -- overwrite existing files
PARALLEL ON       -- write one file per slice (default, fast)
MANIFEST;         -- write a manifest listing all output files

-- Output: s3://my-data-lake/exports/sales_summary/0000_part_00.parquet
--         s3://my-data-lake/exports/sales_summary/0001_part_00.parquet
--         s3://my-data-lake/exports/sales_summary/manifest (manifest JSON)

-- Export to a single CSV (slow on large tables — only use for small result sets)
UNLOAD ('SELECT * FROM analytics.dim_product')
TO 's3://my-data-lake/exports/dim_product/product_export_'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
DELIMITER ','
ADDQUOTES          -- wrap string values in quotes
HEADER             -- add column header row
GZIP               -- compress output
PARALLEL OFF       -- write exactly ONE file (slow — only for small tables)
ALLOWOVERWRITE;

-- Output: s3://my-data-lake/exports/dim_product/product_export_000.gz

CREATE TABLE analytics.pipeline_audit (
    run_id        VARCHAR(64),
    pipeline_name VARCHAR(128),
    status        VARCHAR(20),
    row_count     BIGINT,
    run_date      DATE
)
DISTSTYLE EVEN;   -- round-robin; good for tables not joined on a key

-- Fact table distributed on customer_id
CREATE TABLE analytics.fact_orders (
    order_id    BIGINT,
    customer_id BIGINT,
    product_id  BIGINT,
    order_date  DATE,
    amount      DECIMAL(18,2)
)
DISTSTYLE KEY
DISTKEY (customer_id)        -- distribute by customer_id
SORTKEY (order_date);        -- sort within each slice by date

-- Dimension table also distributed on customer_id
CREATE TABLE analytics.dim_customer (
    customer_id   BIGINT,
    customer_name VARCHAR(200),
    country       VARCHAR(50)
)
DISTSTYLE KEY
DISTKEY (customer_id);   -- same key → co-located with fact_orders

-- This join requires ZERO network redistribution → very fast
SELECT  c.country, SUM(o.amount)
FROM    analytics.fact_orders o
JOIN    analytics.dim_customer c USING (customer_id)
GROUP BY 1;

-- Small lookup table — copy to ALL nodes for fast local joins
CREATE TABLE analytics.dim_country (
    country_code CHAR(2),
    country_name VARCHAR(100),
    region       VARCHAR(50)
)
DISTSTYLE ALL;   -- full copy on every node; tiny table, updated rarely

CREATE TABLE analytics.fact_orders (
    order_id    BIGINT,
    customer_id BIGINT,
    order_date  DATE,
    region      VARCHAR(50),
    amount      DECIMAL(18,2)
)
DISTSTYLE KEY
DISTKEY (customer_id)
COMPOUND SORTKEY (order_date, region);
-- Queries filtering on order_date benefit most
-- Queries filtering on order_date AND region benefit even more
-- Queries filtering on ONLY region (not order_date) get no benefit

CREATE TABLE analytics.fact_events (
    event_id    BIGINT,
    user_id     BIGINT,
    event_date  DATE,
    event_type  VARCHAR(50)
)
INTERLEAVED SORTKEY (user_id, event_date, event_type);
-- Any single-column filter benefits equally
-- But VACUUM takes 3-5× longer — use sparingly

-- Re-sort unsorted rows and reclaim space from deletes
VACUUM analytics.fact_orders;

-- VACUUM only the unsorted region (faster — doesn't re-sort already-sorted rows)
VACUUM SORT ONLY analytics.fact_orders;

-- VACUUM only reclaim space without re-sorting
VACUUM DELETE ONLY analytics.fact_orders;

-- Update query planner statistics (always run after VACUUM)
ANALYZE analytics.fact_orders;

-- Check unsorted percentage before deciding to VACUUM
SELECT "table", unsorted, stats_off, size
FROM   svv_table_info
WHERE  "table" = 'fact_orders';

-- Step 1: Create an external schema backed by Glue Catalog
CREATE EXTERNAL SCHEMA ext_silver
FROM DATA CATALOG
DATABASE 'silver_db'                   -- Glue Catalog database name
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftSpectrumRole'
CREATE EXTERNAL DATABASE IF NOT EXISTS;

-- Step 2: Now query Glue Catalog tables from Redshift directly
-- (tables defined in Glue appear automatically in ext_silver schema)
SELECT  COUNT(*), order_date
FROM    ext_silver.orders          -- this table lives in S3, not Redshift!
WHERE   year = '2024'             -- partition pruning — only reads 2024 data
GROUP BY order_date;

-- Step 3: Join S3 data with internal Redshift table in ONE query
SELECT  c.customer_name,
        SUM(o.amount) AS total_spend
FROM    ext_silver.orders     o    -- S3 via Spectrum
JOIN    analytics.dim_customer c    -- internal Redshift table
  ON  o.customer_id = c.customer_id
GROUP BY 1
ORDER BY 2 DESC
LIMIT    20;

-- See current WLM configuration
SELECT * FROM svl_wlm_query_state;

-- Check queue wait times (is a queue a bottleneck?)
SELECT  service_class,
        COUNT(*) AS queued_queries,
        AVG(queue_start_time) AS avg_wait
FROM    stl_wlm_query
WHERE   queue_start_time > GETDATE() - INTERVAL '1 hour'
GROUP BY service_class;

-- Set query priority for a session (automatic WLM only)
SET query_group TO 'etl_low_priority';

-- Create a materialized view of daily sales (expensive to compute live)
CREATE MATERIALIZED VIEW analytics.mv_daily_sales AS
SELECT  order_date,
        region,
        product_category,
        SUM(amount)    AS total_amount,
        COUNT(DISTINCT customer_id) AS unique_customers,
        COUNT(*)       AS order_count
FROM    analytics.fact_orders
GROUP BY 1, 2, 3;

-- Dashboard queries now hit the pre-built MV — instant results
SELECT * FROM analytics.mv_daily_sales
WHERE  order_date >= CURRENT_DATE - 30;

-- Refresh the MV after each ETL load (or on a schedule)
REFRESH MATERIALIZED VIEW analytics.mv_daily_sales;

-- Auto-refresh option (Redshift refreshes automatically when base table changes)
ALTER MATERIALIZED VIEW analytics.mv_daily_sales AUTO REFRESH YES;

BEGIN;

-- Delete today's partition from target (idempotent — safe to re-run)
DELETE FROM analytics.fact_orders
WHERE  order_date = '2024-06-15';

-- Load fresh data from S3 (today's Parquet partition)
COPY analytics.fact_orders
FROM 's3://my-data-lake/silver/orders/year=2024/month=06/day=15/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET;

COMMIT;   -- atomic: both succeed or both rolled back

-- Step 1: COPY incoming data into a staging table
CREATE TEMP TABLE stg_orders (LIKE analytics.fact_orders);

COPY stg_orders
FROM 's3://my-data-lake/silver/orders/incremental/2024-06-15/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET;

BEGIN;

-- Step 2: Delete existing rows that match incoming keys
DELETE FROM analytics.fact_orders
USING  stg_orders
WHERE  analytics.fact_orders.order_id = stg_orders.order_id;

-- Step 3: Insert all rows from staging (includes new + updated)
INSERT INTO analytics.fact_orders
SELECT * FROM stg_orders;

COMMIT;

-- Staging table auto-dropped when session ends (TEMP table)

import boto3, time, json
from pyspark.sql import SparkSession
from pyspark.sql.functions import lit, current_date

spark = SparkSession.builder.appName("OrdersPipeline").getOrCreate()

# ── 1. Read from source and transform ──────────────────────────
df = spark.read.parquet("s3://my-lake/bronze/orders/date=2024-06-15/")
df_clean = df.filter(df.amount > 0) \
             .withColumn("load_date", current_date()) \
             .dropDuplicates(["order_id"])

# ── 2. Write to S3 Silver as Parquet ───────────────────────────
s3_path = "s3://my-lake/silver/orders/date=2024-06-15/"
df_clean.write.mode("overwrite").parquet(s3_path)
print(f"Wrote {df_clean.count()} rows to S3")

# ── 3. Load from S3 into Redshift via Data API ─────────────────
sm = boto3.client("secretsmanager")
creds = json.loads(sm.get_secret_value(SecretId="prod/redshift/etl")["SecretString"])

rdclient = boto3.client("redshift-data", region_name="us-east-1")

copy_sql = f"""
BEGIN;
DELETE FROM analytics.fact_orders WHERE order_date = '2024-06-15';
COPY analytics.fact_orders
FROM '{s3_path}'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftS3ReadRole'
FORMAT AS PARQUET;
COMMIT;
"""

resp = rdclient.execute_statement(
    ClusterIdentifier="prod-redshift",
    Database="analytics",
    DbUser=creds["username"],
    Sql=copy_sql
)
stmt_id = resp["Id"]

# ── 4. Poll until complete ─────────────────────────────────────
for _ in range(60):
    status = rdclient.describe_statement(Id=stmt_id)["Status"]
    if status in ("FINISHED", "FAILED", "ABORTED"):
        break
    time.sleep(10)

print(f"Redshift COPY status: {status}")
if status != "FINISHED":
    raise RuntimeError("Redshift COPY failed — check stl_load_errors")

from kafka.admin import KafkaAdminClient, NewTopic

admin = KafkaAdminClient(
    bootstrap_servers="b-1.my-msk.abc.kafka.us-east-1.amazonaws.com:9092,"
                      "b-2.my-msk.abc.kafka.us-east-1.amazonaws.com:9092",
    client_id="topic-creator"
)

topic = NewTopic(
    name="prod.orders",
    num_partitions=12,         # 12 partitions = 12 parallel Spark tasks
    replication_factor=3,      # 1 leader + 2 followers per partition
    topic_configs={
        "retention.ms":    "604800000",  # 7 days in ms
        "compression.type":"lz4",        # broker-side compression
        "min.insync.replicas": "2"         # need 2 ISR to accept writes
    }
)

admin.create_topics([topic])
print("Topic prod.orders created with 12 partitions")

import boto3, json
from kafka import KafkaProducer
from aws_schema_registry import SchemaRegistryClient
from aws_schema_registry.avro import AvroSchema
from aws_schema_registry.serde import KafkaSerializer

# ── Define the Avro schema ─────────────────────────────────────
ORDER_SCHEMA = AvroSchema("""{
  "type": "record",
  "name": "Order",
  "namespace": "com.mycompany.events",
  "fields": [
    {"name": "order_id",    "type": "long"},
    {"name": "customer_id", "type": "long"},
    {"name": "amount",      "type": "double"},
    {"name": "order_date",  "type": "string"},
    {"name": "region",      "type": ["null", "string"], "default": null}
  ]
}""")

# ── Create Glue Schema Registry client ───────────────────────
glue_client = boto3.client("glue", region_name="us-east-1")
registry_client = SchemaRegistryClient(glue_client, registry_name="prod-registry")
serializer = KafkaSerializer(registry_client)

# ── Kafka producer with Avro serialiser ──────────────────────
producer = KafkaProducer(
    bootstrap_servers="b-1.my-msk.abc.kafka.us-east-1.amazonaws.com:9092",
    value_serializer=lambda v: serializer.serialize("prod.orders", v, ORDER_SCHEMA)
)

order_event = {
    "order_id":    1001,
    "customer_id": 42,
    "amount":      299.99,
    "order_date":  "2024-06-15",
    "region":      "us-east"
}

producer.send("prod.orders", key=str(order_event["order_id"]).encode(), value=order_event)
producer.flush()
print("Order event published with Avro schema")

from kafka import KafkaProducer
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider

def oauth_cb(oauth_config):
    """Called by kafka-python to get a fresh IAM token."""
    auth_token, expiry_ms = MSKAuthTokenProvider.generate_auth_token("us-east-1")
    return auth_token, expiry_ms / 1000

producer = KafkaProducer(
    bootstrap_servers=[
        "b-1.my-msk.abc.kafka.us-east-1.amazonaws.com:9098",   # IAM port = 9098
        "b-2.my-msk.abc.kafka.us-east-1.amazonaws.com:9098"
    ],
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    sasl_oauth_token_provider=oauth_cb,
    value_serializer=lambda v: v.encode("utf-8")
)

producer.send("prod.orders", value='{"order_id": 1001}')
producer.flush()
print("Message sent with IAM auth")

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "prod.orders",
    bootstrap_servers=[
        "b-1.my-msk.abc.kafka.us-east-1.amazonaws.com:9096"  # SASL_SSL port
    ],
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",
    sasl_plain_username="kafka-user",
    sasl_plain_password="kafka-password",       # fetched from Secrets Manager
    ssl_cafile="/etc/kafka/amazon-root-ca.pem",   # AWS CA certificate
    group_id="spark-etl-cg",
    auto_offset_reset="earliest",
    value_deserializer=lambda m: json.loads(m.decode("utf-8"))
)

for msg in consumer:
    print(f"Partition={msg.partition} Offset={msg.offset} Value={msg.value}")

import boto3
from datetime import datetime, timedelta

cw = boto3.client("cloudwatch", region_name="us-east-1")

# MSK publishes SumOffsetLag metric per consumer group + topic
response = cw.get_metric_statistics(
    Namespace="AWS/Kafka",
    MetricName="SumOffsetLag",
    Dimensions=[
        {"Name": "Cluster Name",     "Value": "prod-msk-cluster"},
        {"Name": "Consumer Group",   "Value": "spark-etl-cg"},
        {"Name": "Topic",            "Value": "prod.orders"}
    ],
    StartTime=datetime.utcnow() - timedelta(minutes=30),
    EndTime=datetime.utcnow(),
    Period=60,
    Statistics=["Maximum"]
)

for dp in sorted(response["Datapoints"], key=lambda x: x["Timestamp"]):
    print(f"{dp['Timestamp'].strftime('%H:%M')}  lag={dp['Maximum']:,.0f}")

# ── Create a CloudWatch alarm if lag exceeds 100k messages ─────
cw.put_metric_alarm(
    AlarmName="MSK-spark-etl-lag-high",
    MetricName="SumOffsetLag",
    Namespace="AWS/Kafka",
    Dimensions=[
        {"Name": "Cluster Name",   "Value": "prod-msk-cluster"},
        {"Name": "Consumer Group", "Value": "spark-etl-cg"},
        {"Name": "Topic",          "Value": "prod.orders"}
    ],
    Statistic="Maximum",
    Period=300,
    EvaluationPeriods=2,
    Threshold=100000,
    ComparisonOperator="GreaterThanThreshold",
    AlarmActions=["arn:aws:sns:us-east-1:123456789012:data-eng-alerts"],
    TreatMissingData="notBreaching"
)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import StructType, StructField, LongType, DoubleType, StringType

spark = SparkSession.builder \
    .appName("MSKOrdersConsumer") \
    .config("spark.sql.shuffle.partitions", "12") \
    .getOrCreate()

# ── Schema for the JSON payload inside Kafka value ─────────────
order_schema = StructType([
    StructField("order_id",    LongType(),   nullable=False),
    StructField("customer_id", LongType(),   nullable=False),
    StructField("amount",      DoubleType(), nullable=True),
    StructField("order_date",  StringType(), nullable=True),
    StructField("region",      StringType(), nullable=True)
])

MSK_BROKERS = ("b-1.my-msk.abc.kafka.us-east-1.amazonaws.com:9098,"
               "b-2.my-msk.abc.kafka.us-east-1.amazonaws.com:9098")

# ── Read stream from MSK ───────────────────────────────────────
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers",           MSK_BROKERS) \
    .option("kafka.security.protocol",            "SASL_SSL") \
    .option("kafka.sasl.mechanism",               "AWS_MSK_IAM") \
    .option("kafka.sasl.jaas.config",
            "software.amazon.msk.auth.iam.IAMLoginModule required;") \
    .option("kafka.sasl.client.callback.handler.class",
            "software.amazon.msk.auth.iam.IAMClientCallbackHandler") \
    .option("subscribe",             "prod.orders") \
    .option("startingOffsets",       "latest") \
    .option("maxOffsetsPerTrigger",  50000) \    # backpressure control
    .load()

# ── Parse JSON value column ────────────────────────────────────
orders = raw_stream \
    .select(
        col("key").cast("string").alias("msg_key"),
        from_json(col("value").cast("string"), order_schema).alias("data"),
        col("partition"),
        col("offset"),
        col("timestamp").alias("kafka_timestamp")
    ) \
    .select("data.*", "partition", "offset", "kafka_timestamp") \
    .withColumn("processed_at", current_timestamp())

# ── Write to Delta Lake (S3) ───────────────────────────────────
query = orders.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://my-lake/checkpoints/orders-msk/") \
    .option("path",              "s3://my-lake/bronze/orders/") \
    .trigger(processingTime="60 seconds") \
    .start()

query.awaitTermination()

from pyspark.sql.functions import to_json, struct

# Enrich the orders stream — join with static product dimension
dim_product = spark.read.parquet("s3://my-lake/silver/dim_product/")
enriched = orders.join(dim_product, "product_id", "left")

# Prepare Kafka output: must have 'value' column as bytes/string
kafka_output = enriched.select(
    col("order_id").cast("string").alias("key"),
    to_json(struct("*")).alias("value")
)

# Write enriched events to a new MSK topic
kafka_output.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", MSK_BROKERS) \
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism",    "AWS_MSK_IAM") \
    .option("kafka.sasl.jaas.config",
            "software.amazon.msk.auth.iam.IAMLoginModule required;") \
    .option("kafka.sasl.client.callback.handler.class",
            "software.amazon.msk.auth.iam.IAMClientCallbackHandler") \
    .option("topic",             "prod.orders-enriched") \
    .option("checkpointLocation", "s3://my-lake/checkpoints/orders-enriched/") \
    .outputMode("append") \
    .trigger(processingTime="60 seconds") \
    .start() \
    .awaitTermination()

{
  "connector.class":                  "io.confluent.connect.s3.S3SinkConnector",
  "tasks.max":                        "6",
  "topics":                           "prod.orders,prod.clicks",
  "s3.region":                        "us-east-1",
  "s3.bucket.name":                   "my-data-lake",
  "s3.part.size":                     "67108864",
  "topics.dir":                       "bronze",
  "flush.size":                       "100000",
  "rotate.interval.ms":               "300000",
  "storage.class":                    "io.confluent.connect.s3.storage.S3Storage",
  "format.class":                     "io.confluent.connect.s3.format.parquet.ParquetFormat",
  "parquet.codec":                    "snappy",
  "partitioner.class":               "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
  "path.format":                      "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
  "timestamp.extractor":             "RecordField",
  "timestamp.field":                 "order_date",
  "schema.compatibility":            "FULL"
}
// Writes: s3://my-data-lake/bronze/prod.orders/year=2024/month=06/day=15/hour=10/
//         part-000001.snappy.parquet
// Every 100k messages OR every 5 minutes — whichever comes first

{
  "connector.class":              "io.confluent.connect.jdbc.JdbcSourceConnector",
  "tasks.max":                    "3",
  "connection.url":               "jdbc:postgresql://my-rds.abc.us-east-1.rds.amazonaws.com:5432/mydb",
  "connection.user":              "kafka_reader",
  "connection.password":          "${file:/opt/kafka/secrets.properties:db.password}",
  "mode":                         "timestamp+incrementing",
  "timestamp.column.name":        "updated_at",
  "incrementing.column.name":     "order_id",
  "table.whitelist":              "public.orders,public.customers",
  "topic.prefix":                 "jdbc.rds.",
  "poll.interval.ms":             "60000",
  "batch.max.rows":               "5000",
  "numeric.mapping":              "best_fit"
}
// Publishes to topics: jdbc.rds.orders, jdbc.rds.customers
// Every 60 seconds polls for rows where updated_at > last_poll_time