Glue Job Config · Ingesting Into Data Lake · aws/agent-toolkit-for-aws · Skills Docs

Glue Job Config · Ingesting Into Data Lake · aws/agent-toolkit-for-aws · Skills Docs

aws glue create-job \
  --name "external-import-<source>-<table>" \
  --role "<glue-role-arn>" \
  --command "Name=glueetl,ScriptLocation=s3://<scripts-bucket>/glue-jobs/external-import-<table>.py,PythonVersion=3" \
  --connections "Connections=<glue-connection-name>" \
  --default-arguments '{
    "--datalake-formats": "iceberg",
    "--connection_name": "<glue-connection-name>",
    "--source_table": "<schema>.<table>",
    "--target_table": "<catalog>.<namespace>.<s3-table>",
    "--watermark_column": "<timestamp-column>",
    "--watermark_bucket": "<bucket>",
    "--watermark_key": "watermarks/<table-name>.txt",
    "--conf": "<see iceberg-catalog-config-and-usage.md for S3 Tables or standard Iceberg catalog config>",
    "--enable-glue-datacatalog": "true",
    "--enable-metrics": "true",
    "--enable-continuous-cloudwatch-log": "true"
  }' \
  --glue-version "5.1" \
  --number-of-workers 5 \
  --worker-type "G.1X" \
  --timeout 60 \
  --max-retries 1 \
  --region <region>

Worker Type	vCPUs	Memory	Use Case
G.1X	4	16 GB	Standard ETL, small to medium data volumes
G.2X	8	32 GB	Large data volumes, memory-intensive transforms
G.4X	16	64 GB	Very large data volumes, complex joins
G.8X	32	128 GB	Massive data volumes, high parallelism

# Read with parallel partitions
source_df = spark.read.format("jdbc").options(
    url=jdbc_url,
    dbtable="large_table",
    numPartitions=10,  # Read with 10 parallel connections
    partitionColumn="id",  # Partition on this column
    lowerBound=1,  # Min value
    upperBound=10000000  # Max value
).load()

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 = source_df.withColumn("row_num", row_number().over(window)) \
    .filter(col("row_num") == 1) \
    .drop("row_num")

from pyspark.sql.functions import col, when

transformed_df = source_df.select(
    # Safe type casting with null handling
    when(col("amount").cast("double").isNotNull(), col("amount").cast("double"))
        .otherwise(0.0).alias("amount"),

    # String trimming and validation
    when(col("email").rlike(r"^[\w\.-]+@[\w\.-]+\.\w+$"), col("email"))
        .otherwise(None).alias("email"),

    # Date parsing with fallback
    when(col("order_date").isNotNull(),
         to_date(col("order_date"), "yyyy-MM-dd"))
        .otherwise(None).alias("order_date")
)

from datetime import timedelta

# Load data from 1 day before last watermark to catch late arrivals
buffer_watermark = (datetime.strptime(last_watermark, '%Y-%m-%d %H:%M:%S')
                    - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')

filtered_df = source_df.filter(
    f"{args['watermark_column']} > '{buffer_watermark}'"
)

# Then use upsert to avoid duplicates

# Install dependencies
pip install pyspark boto3

# Run script locally (modify to use local Spark)
python external-import-customers.py \
  --JOB_NAME test-run \
  --connection_name test-connection \
  --source_table customers \
  --target_table local.test.customers \
  --watermark_column updated_at \
  --watermark_bucket test-bucket \
  --watermark_key watermarks/customers.txt