Incremental Loading

I found these potential watermark columns:
1. CREATED_DATE (TIMESTAMP) - Never changes once set
2. UPDATED_AT (TIMESTAMP) - Updates when record changes (recommended)
3. ID (NUMBER) - Auto-incrementing primary key

Which should I use to track new/updated records?

# Filter for new records
new_records_df = source_df.filter(
    f"{watermark_column} > '{last_watermark}'"
)

# Append to target
new_records_df.writeTo(target_table).append()

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

# Merge into target (upsert)
spark.sql(f"""
MERGE INTO {target_table} AS target
USING changed_records AS source
ON target.{primary_key} = source.{primary_key}
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

# Read all records
all_records_df = source_df.select("*")

# Overwrite target table
all_records_df.writeTo(target_table).overwritePartitions()

import boto3

s3 = boto3.client('s3')
watermark_bucket = args['watermark_bucket']
watermark_key = args['watermark_key']

# Read last watermark
try:
    obj = s3.get_object(Bucket=watermark_bucket, Key=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
    print("No previous watermark found, starting from beginning")

# After loading, update watermark
new_watermark = filtered_df.agg({watermark_column: "max"}).collect()[0][0]
s3.put_object(
    Bucket=watermark_bucket,
    Key=watermark_key,
    Body=str(new_watermark)
)
print(f"Updated watermark to: {new_watermark}")

s3://my-glue-watermarks/
  customers.txt          → "2024-03-12 14:30:00"
  orders.txt             → "2024-03-12 14:25:00"
  products.txt           → "2024-03-10 08:00:00"

aws dynamodb create-table \
  --table-name glue-job-watermarks \
  --attribute-definitions \
    AttributeName=job_name,AttributeType=S \
  --key-schema \
    AttributeName=job_name,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --region <region>

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('glue-job-watermarks')
job_name = args['JOB_NAME']

# Read last watermark
try:
    response = table.get_item(Key={'job_name': job_name})
    item = response['Item']
    last_watermark = item['watermark']
    print(f"Last watermark for {job_name}: {last_watermark}")
except KeyError:
    last_watermark = '1970-01-01 00:00:00'
    print("No previous watermark found, starting from beginning")

# After loading, update watermark
new_watermark = filtered_df.agg({watermark_column: "max"}).collect()[0][0]
table.put_item(Item={
    'job_name': job_name,
    'watermark': str(new_watermark),
    'last_run_time': datetime.now().isoformat(),
    'rows_loaded': row_count
})
print(f"Updated watermark to: {new_watermark}")

# Query target table for max watermark
try:
    max_watermark_df = spark.sql(f"""
        SELECT MAX({watermark_column}) as max_value
        FROM {target_table}
    """)
    last_watermark = max_watermark_df.collect()[0]['max_value']
    if last_watermark is None:
        last_watermark = '1970-01-01 00:00:00'
    print(f"Max watermark in target: {last_watermark}")
except:
    last_watermark = '1970-01-01 00:00:00'
    print("Target table empty or doesn't exist, starting from beginning")

from pyspark.sql.functions import to_utc_timestamp

# Convert source timestamp to UTC
df_utc = source_df.withColumn(
    "timestamp_utc",
    to_utc_timestamp(col("source_timestamp"), "America/New_York")
)

# Batch 1: Load 2020 data
WHERE timestamp >= '2020-01-01' AND timestamp < '2021-01-01'

# Batch 2: Load 2021 data
WHERE timestamp >= '2021-01-01' AND timestamp < '2022-01-01'

# Batch 3: Load 2022+ data
WHERE timestamp >= '2022-01-01'

# Then switch to incremental

# Load data from 1 day before last watermark to catch late arrivals
buffer_watermark = last_watermark - timedelta(days=1)
WHERE timestamp > buffer_watermark

# Use updated_at instead of event_timestamp
WHERE updated_at > last_watermark

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# Deduplicate by primary key, keeping latest by watermark
window = Window.partitionBy("primary_key").orderBy(col(watermark_column).desc())
deduplicated_df = df.withColumn("row_num", row_number().over(window)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

-- Oracle
CREATE INDEX idx_customers_updated_at ON CUSTOMERS(UPDATED_AT);

-- SQL Server
CREATE INDEX idx_customers_updated_at ON CUSTOMERS(UPDATED_AT);

-- PostgreSQL
CREATE INDEX idx_customers_updated_at ON customers(updated_at);

# Load 1 hour of data at a time
batch_size = timedelta(hours=1)
current_watermark = last_watermark

while current_watermark < datetime.now():
    next_watermark = current_watermark + batch_size

    batch_df = source_df.filter(
        (col(watermark_column) > current_watermark) &
        (col(watermark_column) <= next_watermark)
    )

    batch_df.writeTo(target_table).append()

    current_watermark = next_watermark

source_df = spark.read.format("jdbc").options(
    url=jdbc_url,
    dbtable=table_name,
    numPartitions=10,  # Read in parallel with 10 partitions
    partitionColumn=watermark_column,
    lowerBound=last_watermark,
    upperBound=current_time
).load()

# Log metrics
print(f"Job metrics:")
print(f"  Rows loaded: {row_count}")
print(f"  Previous watermark: {last_watermark}")
print(f"  New watermark: {new_watermark}")
print(f"  Watermark advancement: {new_watermark - last_watermark}")
print(f"  Load duration: {load_duration} seconds")

# Publish to CloudWatch (optional)
cloudwatch = boto3.client('cloudwatch')
cloudwatch.put_metric_data(
    Namespace='GlueJobs',
    MetricData=[{
        'MetricName': 'RowsLoaded',
        'Value': row_count,
        'Unit': 'Count',
        'Dimensions': [{'Name': 'JobName', 'Value': job_name}]
    }]
)