Ingesting Into Data Lake

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
import boto3
from datetime import datetime
from pyspark.sql.functions import lit

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table',
    'watermark_column',
    'watermark_bucket',
    'watermark_key'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read last watermark from S3
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'  # Default for timestamp
    # OR last_watermark = '0'  # Default for ID column
    print("No previous watermark found, starting from beginning")

# Read from external database using Glue connection
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

# Apply incremental filter
filtered_df = source_df.filter(
    f"{args['watermark_column']} > '{last_watermark}'"
)

row_count = filtered_df.count()
print(f"Loading {row_count} new/updated records")

if row_count > 0:
    # Apply transformations (type casting, column mapping, etc.)
    transformed_df = filtered_df.select(
        # Map source columns to target schema
        filtered_df["source_col1"].cast("int").alias("target_col1"),
        filtered_df["source_col2"].alias("target_col2"),
        filtered_df["source_col3"].cast("double").alias("target_col3"),
        # Add load metadata
        lit(datetime.now()).alias("load_timestamp")
    )

    # Write to Iceberg table (append mode)
    transformed_df.writeTo(args['target_table']).append()

    # Update watermark in S3
    new_watermark = filtered_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
    print(f"Successfully loaded {row_count} records")
else:
    print("No new records to load")

job.commit()

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col, lit
import boto3
from datetime import datetime

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table',
    'watermark_column',
    'primary_key',  # Column used for merging
    'watermark_bucket',
    'watermark_key'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read last watermark
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'
    print("No previous watermark found, starting from beginning")

# Read from external database
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

# Get new/updated records
changed_records_df = source_df.filter(
    f"{args['watermark_column']} > '{last_watermark}'"
)

row_count = changed_records_df.count()
print(f"Found {row_count} new/updated records")

if row_count > 0:
    # Apply transformations
    transformed_df = changed_records_df.select(
        changed_records_df["customer_id"].cast("int").alias("customer_id"),
        changed_records_df["customer_name"].alias("name"),
        changed_records_df["email"].alias("email"),
        changed_records_df["status"].alias("status"),
        changed_records_df["updated_at"].alias("updated_at"),
        lit(datetime.now()).alias("load_timestamp")
    )

    # Create temporary view for MERGE operation
    transformed_df.createOrReplaceTempView("source_view")

    # Execute MERGE INTO (upsert)
    spark.sql(f"""
    MERGE INTO {args['target_table']} AS target
    USING source_view AS source
    ON target.{args['primary_key']} = source.{args['primary_key']}
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
    """)

    # Update watermark
    new_watermark = changed_records_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
    print(f"Upserted {row_count} records")
else:
    print("No new records to process")

job.commit()

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

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_query',  # SQL query to execute
    'target_table',
    'watermark_column',
    'watermark_bucket',
    'watermark_key',
    'jdbc_driver'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Retrieve JDBC credentials from Glue connection
jdbc_conf = glueContext.extract_jdbc_conf(args['connection_name'])

# Read last watermark
s3 = boto3.client('s3')
try:
    obj = s3.get_object(Bucket=args['watermark_bucket'], Key=args['watermark_key'])
    last_watermark = obj['Body'].read().decode('utf-8').strip()
    print(f"Last watermark: {last_watermark}")
except s3.exceptions.NoSuchKey:
    last_watermark = '1970-01-01 00:00:00'
    print("Starting from beginning")

# Build query with watermark filter
query = f"""
SELECT * FROM ({args['source_query']}) AS base_query
WHERE {args['watermark_column']} > '{last_watermark}'
"""

print(f"Executing query: {query}")

# Read using JDBC with custom query
source_df = spark.read.format("jdbc").options(
    url=jdbc_conf['url'],
    dbtable=f"({query}) AS subquery",
    user=jdbc_conf['user'],
    password=jdbc_conf['password'],
    driver=args['jdbc_driver']  # e.g., "oracle.jdbc.OracleDriver"
).load()

row_count = source_df.count()
print(f"Query returned {row_count} records")

if row_count > 0:
    # Add load metadata
    transformed_df = source_df.withColumn("load_timestamp", lit(datetime.now()))

    # Write to Iceberg table
    transformed_df.writeTo(args['target_table']).append()

    # Update watermark
    new_watermark = source_df.agg({args['watermark_column']: "max"}).collect()[0][0]
    s3.put_object(
        Bucket=args['watermark_bucket'],
        Key=args['watermark_key'],
        Body=str(new_watermark)
    )
    print(f"Updated watermark to: {new_watermark}")
else:
    print("No new records")

job.commit()

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import lit
from datetime import datetime

# Parse job arguments
args = getResolvedOptions(sys.argv, [
    'JOB_NAME',
    'connection_name',
    'source_table',
    'target_table'
])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read all records from source
source_df = glueContext.create_dynamic_frame.from_catalog(
    database="<temp-catalog-db>",
    table_name="<source-table>",
    transformation_ctx="source_df",
    additional_options={
        "connectionName": args['connection_name']
    }
).toDF()

row_count = source_df.count()
print(f"Loading {row_count} records (full refresh)")

# Apply transformations
transformed_df = source_df.select(
    source_df["col1"].alias("col1"),
    source_df["col2"].alias("col2"),
    lit(datetime.now()).alias("load_timestamp")
)

# Overwrite target table
transformed_df.writeTo(args['target_table']).overwritePartitions()

print(f"Full refresh completed: {row_count} records loaded")

job.commit()