### Install PySpark Locally

Source: https://github.com/awslabs/aws-glue-libs/blob/master/awsglue/README.md

Install PySpark locally to enable auto-completion in an IDE when working with the awsglue library.

```bash
pip install pyspark
```

--------------------------------

### Run Glue Executables

Source: https://github.com/awslabs/aws-glue-libs/blob/master/README.md

Execute the provided scripts to start a Glue shell, submit a Glue Spark application, or run Pytest for your Glue ETL jobs locally.

```bash
Glue shell: ./bin/gluepyspark
```

```bash
Glue submit: ./bin/gluesparksubmit
```

```bash
pytest: ./bin/gluepytest
```

--------------------------------

### GlueContext Initialization

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Demonstrates how to initialize the GlueContext, which serves as the central entry point for AWS Glue ETL jobs, extending PySpark's SQLContext.

```APIDOC
## GlueContext

`GlueContext(sparkContext, **options)` is the primary entry point that wraps PySpark's `SQLContext` and registers all Glue JVM classes. It exposes convenience reader/writer accessors (`create_dynamic_frame`, `write_dynamic_frame`, `create_data_frame`, `write_data_frame`) and is the factory for all data sources and sinks.

### Request Example
```python
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", "TempDir"])

sc = SparkContext()
glue_ctx = GlueContext(sc)
spark = glue_ctx.spark_session

job = Job(glue_ctx)
job.init(args["JOB_NAME"], args)

# GlueContext is now ready; access logger
logger = glue_ctx.get_logger()
logger.info("GlueContext initialized successfully")

job.commit()
```
```

--------------------------------

### Initialize GlueContext and Job

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Sets up the GlueContext, SparkSession, and Job objects, which are central to AWS Glue ETL jobs. Ensures job parameters are resolved and logger is accessible.

```python
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", "TempDir"])

sc = SparkContext()
glue_ctx = GlueContext(sc)
spark = glue_ctx.spark_session

job = Job(glue_ctx)
job.init(args["JOB_NAME"], args)

# GlueContext is now ready; access logger
logger = glue_ctx.get_logger()
logger.info("GlueContext initialized successfully")

job.commit()
```

--------------------------------

### Job Initialization and Commit

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Initializes job bookmark state using Job.init() and persists the bookmark upon successful completion using Job.commit().

```APIDOC
## Job.init / Job.commit

`Job.init(job_name, args)` initializes job bookmark state; `Job.commit()` persists the bookmark so the next run processes only new data. Always call `commit()` at the end of a successful job.

### Request Example
```python
from awsglue.job import Job

job = Job(glue_ctx)
job.init(args["JOB_NAME"], args)

# ... ETL logic ...

# Commit bookmarks only after successful completion
job.commit()
```
```

--------------------------------

### Initialize and Commit Job Bookmarks

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Initializes job bookmark state using Job.init and persists it with Job.commit upon successful completion. This ensures incremental processing of data in subsequent runs.

```python
from awsglue.job import Job

job = Job(glue_ctx)
job.init(args["JOB_NAME"], args)

# ... ETL logic ...

# Commit bookmarks only after successful completion
job.commit()
```

--------------------------------

### Read from a JDBC source

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Demonstrates how to create a DynamicFrame by reading data from a JDBC data source using connection details.

```APIDOC
## Read from a JDBC source
```python
jdbc_dyf = glue_ctx.create_dynamic_frame.from_options(
    connection_type="jdbc",
    connection_options={
        "url": "jdbc:postgresql://host:5432/mydb",
        "dbtable": "public.customers",
        "user": "etl_user",
        "password": "secret"
    },
    transformation_ctx="jdbc_dyf"
)
```
```

--------------------------------

### GlueContext.create_dynamic_frame.from_options

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Creates a DynamicFrame directly from a connection type and options, without requiring a catalog entry. Supports various data sources like S3, JDBC, DynamoDB, Kinesis, and Kafka.

```APIDOC
## GlueContext.create_dynamic_frame.from_options

Creates a `DynamicFrame` directly from a connection type and options without needing a catalog entry. Supports `s3`, `jdbc`, `dynamodb`, `kinesis`, `kafka`, and more.

### Request Example
```python
# Read JSON files from S3
raw_dyf = glue_ctx.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={
        "paths": ["s3://my-bucket/raw/events/"],
        "recurse": True
    },
    format="json",
    format_options={"jsonPath": "$.records[*]"},
    transformation_ctx="raw_dyf"
)
```
```

--------------------------------

### Resolve Job Options with getResolvedOptions

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Parses job parameters from sys.argv, handling Glue-specific arguments and making them available for job logic. Use this to access parameters like JOB_NAME or TempDir.

```python
import sys
from awsglue.utils import getResolvedOptions

# argv example: script.py --JOB_NAME my-job --TempDir s3://my-bucket/tmp --source_db mydb
args = getResolvedOptions(
    sys.argv,
    ["JOB_NAME", "TempDir", "source_db", "target_path"]
)

print(args["JOB_NAME"])	# "my-job"
print(args["TempDir"])	# "s3://my-bucket/tmp"
print(args["source_db"])	# "mydb"
print(args["target_path"])	# resolved value
```

--------------------------------

### Partition Fields with SplitFields

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use `SplitFields.apply` to partition columns into two `DynamicFrame`s: one with specified fields and another with the remainder.

```python
from awsglue.transforms import SplitFields, SelectFromCollection

dfc = SplitFields.apply(
    frame=orders_dyf,
    paths=["customer_id", "email", "phone"],
    name1="pii_frame",
    name2="order_data_frame",
    transformation_ctx="split_pii"
)

pii_dyf       = SelectFromCollection.apply(dfc, "pii_frame", "get_pii")
order_data_dyf = SelectFromCollection.apply(dfc, "order_data_frame", "get_orders")

```

--------------------------------

### GlueContext.create_dynamic_frame.from_catalog

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Reads a table from the AWS Glue Data Catalog into a DynamicFrame, supporting partition pruning and additional options for features like bookmarking.

```APIDOC
## GlueContext.create_dynamic_frame.from_catalog

Reads a table registered in the AWS Glue Data Catalog into a `DynamicFrame`. Supports push-down predicates for partition pruning and `additional_options` for features like bookmarking and DynamoDB export reads.

### Request Example
```python
# Read a partitioned S3 table, pruning to a single day
orders_dyf = glue_ctx.create_dynamic_frame.from_catalog(
    database="ecommerce",
    table_name="orders_parquet",
    push_down_predicate="order_date = '2024-01-15'",
    transformation_ctx="orders_dyf"
)

orders_dyf.printSchema()
# root
# |-- order_id: long
# |-- customer_id: long
# |-- order_date: string
# |-- items: array
#     |-- element: struct
#         |-- sku: string
#         |-- qty: int

print(f"Record count: {orders_dyf.count()}")
orders_dyf.show(5)
```
```

--------------------------------

### getResolvedOptions

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Parses job parameters from sys.argv, handling reserved Glue parameters and making them globally accessible.

```APIDOC
## getResolvedOptions

`getResolvedOptions(args, options)` parses job parameters from `sys.argv`, handles reserved Glue parameters (job bookmark, temp dir, encryption), and stores them globally so transforms like `Relationalize` can access `TempDir` automatically.

### Request Example
```python
import sys
from awsglue.utils import getResolvedOptions

# argv example: script.py --JOB_NAME my-job --TempDir s3://my-bucket/tmp --source_db mydb
args = getResolvedOptions(
    sys.argv,
    ["JOB_NAME", "TempDir", "source_db", "target_path"]
)

print(args["JOB_NAME"])
print(args["TempDir"])
print(args["source_db"])
print(args["target_path"])
```
```

--------------------------------

### SelectFromCollection / MapToCollection / FlatMap

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Transformations for `DynamicFrameCollection`. `SelectFromCollection` picks a frame by key, `MapToCollection` applies a function to all frames, and `FlatMap` applies a function and flattens the results.

```APIDOC
## SelectFromCollection / MapToCollection / FlatMap

`SelectFromCollection` picks one `DynamicFrame` from a `DynamicFrameCollection` by key. `MapToCollection` applies a transform to every frame in a collection. `FlatMap` applies a transform that can return one or many frames, then flattens the results.

```python
from awsglue.transforms import SelectFromCollection, MapToCollection, FlatMap, DropNullFields

# After Relationalize, pick the root table
root_dyf = SelectFromCollection.apply(dfc, "orders", "select_root")

# Apply DropNullFields to every frame in the collection
clean_dfc = MapToCollection.apply(
    dfc=dfc,
    callable=lambda frame, ctx: DropNullFields.apply(frame, transformation_ctx=ctx),
    transformation_ctx="clean_collection"
)

# FlatMap: apply ApplyMapping to each frame, potentially splitting into more frames
result_dfc = FlatMap.apply(
    dfc=clean_dfc,
    BaseTransform=DropNullFields,
    frame_name="frame",
    transformation_ctx="flatmap_clean"
)
```
```

--------------------------------

### GlueContext.write_dynamic_frame.from_options

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Writes a DynamicFrame to an external target using specified connection type and format. Supports writing to S3 with various formats and compression, or back to a catalog-defined table.

```APIDOC
## GlueContext.write_dynamic_frame.from_options

Writes a `DynamicFrame` to an external target using specified connection type and format. The `from_catalog` variant writes back to a catalog-defined table location.

```python
# Write to S3 as gzip-compressed Parquet
glu_ctx.write_dynamic_frame.from_options(
    frame=transformed_dyf,
    connection_type="s3",
    connection_options={
        "path": "s3://my-bucket/processed/orders/",
        "partitionKeys": ["order_date"]
    },
    format="glueparquet",
    format_options={"compression": "gzip"},
    transformation_ctx="write_orders"
)

# Write back to catalog table (e.g., Redshift)
glu_ctx.write_dynamic_frame.from_catalog(
    frame=transformed_dyf,
    database="ecommerce",
    table_name="orders_redshift",
    redshift_tmp_dir=args["TempDir"],
    transformation_ctx="write_redshift"
)
```
```

--------------------------------

### Manipulate DynamicFrame Collections

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use SelectFromCollection to pick a DynamicFrame by key, MapToCollection to apply a transform to all frames, and FlatMap to apply a transform that can return multiple frames, flattening the results.

```python
from awsglue.transforms import SelectFromCollection, MapToCollection, FlatMap, DropNullFields

# After Relationalize, pick the root table
root_dyf = SelectFromCollection.apply(dfc, "orders", "select_root")

# Apply DropNullFields to every frame in the collection
clean_dfc = MapToCollection.apply(
    dfc=dfc,
    callable=lambda frame, ctx: DropNullFields.apply(frame, transformation_ctx=ctx),
    transformation_ctx="clean_collection"
)

# FlatMap: apply ApplyMapping to each frame, potentially splitting into more frames
result_dfc = FlatMap.apply(
    dfc=clean_dfc,
    BaseTransform=DropNullFields,
    frame_name="frame",
    transformation_ctx="flatmap_clean"
)
```

--------------------------------

### Write DynamicFrame to Catalog Table (Redshift)

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Writes a DynamicFrame back to a catalog-defined table, suitable for targets like Redshift. Requires a temporary S3 directory for Redshift.

```python
# Write back to catalog table (e.g., Redshift)
glue_ctx.write_dynamic_frame.from_catalog(
    frame=transformed_dyf,
    database="ecommerce",
    table_name="orders_redshift",
    redshift_tmp_dir=args["TempDir"],
    transformation_ctx="write_redshift"
)
```

--------------------------------

### Process Streaming Data with GlueContext.forEachBatch

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Process streaming data in micro-batches using forEachBatch. This method handles checkpointing, retries, and per-batch persistence for fault tolerance.

```python
from awsglue.context import GlueContext

# Read from Kinesis stream
kinesis_df = glue_ctx.create_data_frame.from_options(
    connection_type="kinesis",
    connection_options={
        "typeOfData": "kinesis",
        "streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/events",
        "classification": "json",
        "startingPosition": "TRIM_HORIZON"
    }
)

def process_batch(data_frame, batch_id):
    dyf = DynamicFrame.fromDF(data_frame, glue_ctx, "batch_" + str(batch_id))
    mapped = ApplyMapping.apply(dyf, [("event_type","string","EventType","string")])
    glue_ctx.write_dynamic_frame.from_options(
        frame=mapped,
        connection_type="s3",
        connection_options={"path": "s3://my-bucket/streaming/"},
        format="parquet"
    )

glue_ctx.forEachBatch(
    frame=kinesis_df,
    batch_function=process_batch,
    options={
        "windowSize": "100 seconds",
        "checkpointLocation": "s3://my-bucket/checkpoints/kinesis/",
        "batchMaxRetries": 3
    }
)
```

--------------------------------

### Join.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Performs an equality join between two DynamicFrame objects on the specified key columns.

```APIDOC
## Join.apply

### Description
Performs an equality join between two `DynamicFrame` objects on the specified key columns.

### Method Signature
`Join.apply(frame1, frame2, keys1, keys2)`

### Parameters
- **frame1** (DynamicFrame) - The first input DynamicFrame.
- **frame2** (DynamicFrame) - The second input DynamicFrame.
- **keys1** (list) - A list of key column names from `frame1` to join on.
- **keys2** (list) - A list of key column names from `frame2` to join on.

### Example
```python
from awsglue.transforms import Join

# Join orders with customer dimension
enriched_dyf = Join.apply(
    frame1=orders_dyf,
    frame2=customers_dyf,
    keys1=["customer_id"],
    keys2=["id"],
    transformation_ctx="join_customers"
)

enriched_dyf.printSchema()
```
```

--------------------------------

### DynamicFrame.spigot

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Writes a sample of records to an S3 path mid-pipeline for debugging or auditing without interrupting the main data flow. The original frame is returned unchanged.

```APIDOC
## DynamicFrame.spigot

### Description
Writes a sample of records from a `DynamicFrame` to an S3 path for debugging or auditing purposes. This operation does not alter the data flow, and the original frame is returned unchanged.

### Method Signature
`spigot(path, options)`

### Parameters
- **path** (str) - The S3 path where the sample records will be written.
- **options** (dict) - A dictionary of options to control sampling (e.g., `{"topk": 10}` to get the top 10 records).

### Example
```python
# Write top-10 records to S3 for inspection, then continue pipeline
sampled_dyf = raw_dyf.spigot(
    path="s3://my-bucket/debug/sample/",
    options={"topk": 10},
    transformation_ctx="spigot_debug"
)

# sampled_dyf is the same as raw_dyf — processing continues normally
mapped_dyf = ApplyMapping.apply(frame=sampled_dyf, mappings=[...])
```
```

--------------------------------

### Read Table from Glue Data Catalog

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Creates a DynamicFrame from a table defined in the AWS Glue Data Catalog. Supports partition pruning via push_down_predicate for efficient data loading.

```python
# Read a partitioned S3 table, pruning to a single day
orders_dyf = glue_ctx.create_dynamic_frame.from_catalog(
    database="ecommerce",
    table_name="orders_parquet",
    push_down_predicate="order_date = '2024-01-15'",
    transformation_ctx="orders_dyf"
)

orders_dyf.printSchema()
# root
# |-- order_id: long
# |-- customer_id: long
# |-- order_date: string
# |-- items: array
#     |-- element: struct
#         |-- sku: string
#         |-- qty: int

print(f"Record count: {orders_dyf.count()}")
orders_dyf.show(5)
```

--------------------------------

### Spigot DynamicFrame for Sampling

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Write a sample of records from a DynamicFrame to S3 mid-pipeline for debugging or auditing without interrupting the main data flow. The original frame is returned unchanged.

```python
# Write top-10 records to S3 for inspection, then continue pipeline
sampled_dyf = raw_dyf.spigot(
    path="s3://my-bucket/debug/sample/",
    options={"topk": 10},
    transformation_ctx="spigot_debug"
)

# sampled_dyf is the same as raw_dyf — processing continues normally
mapped_dyf = ApplyMapping.apply(frame=sampled_dyf, mappings=[...])
```

--------------------------------

### ApplyMapping.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Applies column transformations including renaming, type casting, and dropping fields based on a provided list of mappings. It allows for declarative data shaping.

```APIDOC
## ApplyMapping

`ApplyMapping.apply(frame, mappings)` renames columns, casts types, and drops unwanted fields in one declarative step using a list of `(source_col, source_type, target_col, target_type)` tuples.

```python
from awsglue.transforms import ApplyMapping

mapped_dyf = ApplyMapping.apply(
    frame=raw_dyf,
    mappings=[
        ("order_id",    "long",   "OrderId",      "string"),
        ("customer_id", "long",   "CustomerId",   "long"),
        ("order_date",  "string", "OrderDate",    "string"),
        ("total_amt",   "double", "TotalAmount",  "double"),
        # Omitting a source column effectively drops it
    ],
    case_sensitive=False,
    transformation_ctx="mapped_dyf"
)

mapped_dyf.printSchema()
# root
# |-- OrderId: string
# |-- CustomerId: long
# |-- OrderDate: string
# |-- TotalAmount: double
```
```

--------------------------------

### SplitFields.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Partitions the columns of a DynamicFrame into two separate frames based on specified paths.

```APIDOC
## SplitFields.apply

### Description
Partitions the columns of a `DynamicFrame` into two distinct `DynamicFrame`s. One frame (`name1`) contains the fields specified in `paths`, and the other frame (`name2`) contains the remaining fields.

### Parameters
- **frame** (DynamicFrame) - The input DynamicFrame.
- **paths** (list of strings) - A list of column names or nested paths to include in the first output frame.
- **name1** (string) - The name for the frame containing the specified paths.
- **name2** (string) - The name for the frame containing the remaining fields.
- **transformation_ctx** (string) - An identifier for the transformation.

### Request Example
```python
from awsglue.transforms import SplitFields, SelectFromCollection

dfc = SplitFields.apply(
    frame=orders_dyf,
    paths=["customer_id", "email", "phone"],
    name1="pii_frame",
    name2="order_data_frame",
    transformation_ctx="split_pii"
)

pii_dyf       = SelectFromCollection.apply(dfc, "pii_frame", "get_pii")
order_data_dyf = SelectFromCollection.apply(dfc, "order_data_frame", "get_orders")
```
```

--------------------------------

### Write DynamicFrame to S3 as Parquet

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Writes a DynamicFrame to an S3 location in gzip-compressed Parquet format. Supports partitioning by specified keys.

```python
# Write to S3 as gzip-compressed Parquet
glue_ctx.write_dynamic_frame.from_options(
    frame=transformed_dyf,
    connection_type="s3",
    connection_options={
        "path": "s3://my-bucket/processed/orders/",
        "partitionKeys": ["order_date"]
    },
    format="glueparquet",
    format_options={"compression": "gzip"},
    transformation_ctx="write_orders"
)
```

--------------------------------

### GlueContext.forEachBatch (Streaming)

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Process a streaming DataFrame in micro-batches using `forEachBatch`. This method provides fault tolerance through automatic retries and checkpointing.

```APIDOC
## GlueContext.forEachBatch (Streaming)

`forEachBatch(frame, batch_function, options)` processes a streaming `DataFrame` in micro-batches. It handles checkpointing, automatic retries with back-off, and per-batch persistence for fault tolerance.

```python
from awsglue.context import GlueContext

# Read from Kinesis stream
kinesis_df = glue_ctx.create_data_frame.from_options(
    connection_type="kinesis",
    connection_options={
        "typeOfData": "kinesis",
        "streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/events",
        "classification": "json",
        "startingPosition": "TRIM_HORIZON"
    }
)

def process_batch(data_frame, batch_id):
    dyf = DynamicFrame.fromDF(data_frame, glue_ctx, "batch_" + str(batch_id))
    mapped = ApplyMapping.apply(dyf, [("event_type","string","EventType","string")])
    glue_ctx.write_dynamic_frame.from_options(
        frame=mapped,
        connection_type="s3",
        connection_options={"path": "s3://my-bucket/streaming/"},
        format="parquet"
    )

glue_ctx.forEachBatch(
    frame=kinesis_df,
    batch_function=process_batch,
    options={
        "windowSize": "100 seconds",
        "checkpointLocation": "s3://my-bucket/checkpoints/kinesis/",
        "batchMaxRetries": 3
    }
)
```
```

--------------------------------

### Join DynamicFrames on Keys

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Perform an equality join between two DynamicFrames using specified key columns. Be mindful of potential duplicate key columns in the output schema, which are prefixed with '.'

```python
from awsglue.transforms import Join

# Join orders with customer dimension
enriched_dyf = Join.apply(
    frame1=orders_dyf,
    frame2=customers_dyf,
    keys1=["customer_id"],
    keys2=["id"],
    transformation_ctx="join_customers"
)

enriched_dyf.printSchema()
# root
# |-- order_id: long
# |-- customer_id: long
# |-- total_amount: double
# |-- .id: long          ← duplicate key from customers frame
# |-- .name: string
# |-- .email: string
```

--------------------------------

### Export SPARK_HOME for Glue Versions

Source: https://github.com/awslabs/aws-glue-libs/blob/master/README.md

Set the SPARK_HOME environment variable to the extracted location of the Apache Spark distribution corresponding to your AWS Glue version. This is crucial for running Glue Spark code locally.

```bash
Glue version 2.0: export SPARK_HOME=/home/$USER/spark-2.4.3-bin-hadoop2.8
```

```bash
Glue version 3.0: export SPARK_HOME=/home/$USER/spark-3.1.1-amzn-0-bin-3.2.1-amzn-3
```

```bash
Glue version 4.0: export SPARK_HOME=/home/$USER/spark-3.3.0-amzn-1-bin-3.3.3-amzn-0
```

```bash
Glue version 5.0: export SPARK_HOME=/home/$USER/spark-3.5.4-bin-hadoop3
```

```bash
Glue version 5.1: export SPARK_HOME=/home/$USER/spark-3.5.6-bin-hadoop3
```

--------------------------------

### Drop or Select Fields with DropFields/SelectFields

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use `DropFields.apply` to remove specified columns or `SelectFields.apply` to keep only named columns. Supports nested paths.

```python
from awsglue.transforms import DropFields, SelectFields

# Remove sensitive and internal fields
clean_dyf = DropFields.apply(
    frame=raw_dyf,
    paths=["internal_id", "pii.ssn", "audit.created_by"],
    transformation_ctx="drop_fields"
)

# Keep only the columns needed for the output table
slim_dyf = SelectFields.apply(
    frame=clean_dyf,
    paths=["order_id", "customer_id", "total_amount", "order_date"],
    transformation_ctx="select_fields"
)
```

--------------------------------

### Repartition and Coalesce DynamicFrames

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use repartition to increase partitions for parallelism or coalesce to reduce partitions for writing. Coalesce avoids a full shuffle when only reducing partition count.

```python
# Increase parallelism for a large dataset before a heavy transform
large_dyf = raw_dyf.repartition(200, transformation_ctx="repartition_up")

# Reduce to a single file before writing a small reference table
small_dyf = reference_dyf.coalesce(1, shuffle=False, transformation_ctx="coalesce_ref")

glue_ctx.write_dynamic_frame.from_options(
    frame=small_dyf,
    connection_type="s3",
    connection_options={"path": "s3://my-bucket/reference/"},
    format="csv"
)
```

--------------------------------

### DropFields.apply / SelectFields.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

DropFields.apply removes specified columns, while SelectFields.apply keeps only the named columns.

```APIDOC
## DropFields.apply / SelectFields.apply

### Description
`DropFields.apply` removes specified columns (including nested paths) from a `DynamicFrame`. `SelectFields.apply` is the inverse operation, keeping only the specified columns.

### Parameters for DropFields.apply
- **frame** (DynamicFrame) - The input DynamicFrame.
- **paths** (list of strings) - A list of column names or nested paths to drop.
- **transformation_ctx** (string) - An identifier for the transformation.

### Parameters for SelectFields.apply
- **frame** (DynamicFrame) - The input DynamicFrame.
- **paths** (list of strings) - A list of column names or nested paths to keep.
- **transformation_ctx** (string) - An identifier for the transformation.

### Request Example
```python
from awsglue.transforms import DropFields, SelectFields

# Remove sensitive and internal fields
clean_dyf = DropFields.apply(
    frame=raw_dyf,
    paths=["internal_id", "pii.ssn", "audit.created_by"],
    transformation_ctx="drop_fields"
)

# Keep only the columns needed for the output table
slim_dyf = SelectFields.apply(
    frame=clean_dyf,
    paths=["order_id", "customer_id", "total_amount", "order_date"],
    transformation_ctx="select_fields"
)
```
```

--------------------------------

### SplitRows.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Splits records into two frames based on column-level comparison predicates. Records matching all predicates go to name1; the rest go to name2.

```APIDOC
## SplitRows.apply

### Description
Splits records into two frames based on column-level comparison predicates. Records matching all predicates go to `name1`; the rest go to `name2`.

### Method Signature
`SplitRows.apply(frame, comparison_dict, name1, name2)`

### Parameters
- **frame** (DynamicFrame) - The input DynamicFrame.
- **comparison_dict** (dict) - A dictionary defining the comparison predicates.
- **name1** (str) - The name for the frame containing records that match the predicates.
- **name2** (str) - The name for the frame containing records that do not match the predicates.

### Example
```python
from awsglue.transforms import SplitRows, SelectFromCollection

# Separate high-value orders (amount > 1000) from the rest
dfc = SplitRows.apply(
    frame=orders_dyf,
    comparison_dict={"total_amount": ">" : 1000}},
    name1="high_value",
    name2="standard",
    transformation_ctx="split_rows"
)

high_value_dyf = SelectFromCollection.apply(dfc, "high_value")
standard_dyf   = SelectFromCollection.apply(dfc, "standard")
```
```

--------------------------------

### Map.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Applies a transformation function to every DynamicRecord. If the function returns None or raises an exception, the record is marked as an error.

```APIDOC
## Map.apply

### Description
Applies a transformation function `f` to every `DynamicRecord` in the `frame`. Records for which `f` returns `None` or raises an exception are marked as errors.

### Parameters
- **frame** (DynamicFrame) - The input DynamicFrame.
- **f** (function) - The transformation function to apply to each record.
- **transformation_ctx** (string) - An identifier for the transformation.

### Request Example
```python
from awsglue.transforms import Map

def enrich_record(record):
    # Normalize email to lowercase
    if "email" in record:
        record["email"] = record["email"].lower().strip()
    # Compute derived field
    record["full_name"] = f"{record.get('first_name', '')} {record.get('last_name', '')}".strip()
    # Remove PII
    if "ssn" in record:
        del record["ssn"]
    return record

enriched_dyf = Map.apply(
    frame=customers_dyf,
    f=enrich_record,
    transformation_ctx="enriched_dyf"
)
```
```

--------------------------------

### Create DynamicFrame from S3 Options

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Creates a DynamicFrame directly from S3 data without a catalog entry. Specify paths, recursion, and format options like JSONPath for parsing nested structures.

```python
# Read JSON files from S3
raw_dyf = glue_ctx.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={
        "paths": ["s3://my-bucket/raw/events/"]
        "recurse": True
    },
    format="json",
    format_options={"jsonPath": "$.records[*]"},
    transformation_ctx="raw_dyf"
)

```

--------------------------------

### Read from JDBC Source

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Reads data from a JDBC data source into a DynamicFrame. Ensure the JDBC connection details are correctly configured.

```python
jdbc_dyf = glue_ctx.create_dynamic_frame.from_options(
    connection_type="jdbc",
    connection_options={
        "url": "jdbc:postgresql://host:5432/mydb",
        "dbtable": "public.customers",
        "user": "etl_user",
        "password": "secret"
    },
    transformation_ctx="jdbc_dyf"
)
```

--------------------------------

### DynamicFrame.repartition / coalesce

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Control the number of partitions in a DynamicFrame. `repartition` shuffles data into a specified number of partitions, while `coalesce` reduces partitions, optionally with shuffling.

```APIDOC
## DynamicFrame.repartition / coalesce

`repartition(num_partitions)` shuffles data into exactly `num_partitions` partitions; `coalesce(num_partitions, shuffle=False)` reduces partition count with optional shuffle, avoiding a full reshuffle when only reducing.

```python
# Increase parallelism for a large dataset before a heavy transform
large_dyf = raw_dyf.repartition(200, transformation_ctx="repartition_up")

# Reduce to a single file before writing a small reference table
small_dyf = reference_dyf.coalesce(1, shuffle=False, transformation_ctx="coalesce_ref")

glue_ctx.write_dynamic_frame.from_options(
    frame=small_dyf,
    connection_type="s3",
    connection_options={"path": "s3://my-bucket/reference/"},
    format="csv"
)
```
```

--------------------------------

### Apply Mapping Transformation

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Renames columns, casts types, and drops fields from a DynamicFrame using a declarative mapping. Case sensitivity can be controlled.

```python
from awsglue.transforms import ApplyMapping

mapped_dyf = ApplyMapping.apply(
    frame=raw_dyf,
    mappings=[
        ("order_id",    "long",   "OrderId",      "string"),
        ("customer_id", "long",   "CustomerId",   "long"),
        ("order_date",  "string", "OrderDate",    "string"),
        ("total_amt",   "double", "TotalAmount",  "double"),
        # Omitting a source column effectively drops it
    ],
    case_sensitive=False,
    transformation_ctx="mapped_dyf"
)

mapped_dyf.printSchema()
# root
# |-- OrderId: string
# |-- CustomerId: long
# |-- OrderDate: string
# |-- TotalAmount: double
```

--------------------------------

### Write DynamicFrames to S3

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Write DynamicFrames to S3 in specified formats (JSON, Parquet). Ensure the target S3 paths are correctly configured.

```python
glue_ctx.write_dynamic_frame.from_options(pii_dyf,
    "s3",
    {"path": "s3://secure-bucket/pii/"}, "json")
gglue_ctx.write_dynamic_frame.from_options(order_data_dyf, "s3",
    {"path": "s3://analytics-bucket/orders/"}, "parquet")
```

--------------------------------

### Purge Table Data and S3 Paths

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use purge_table to delete S3 files older than a retention period for a catalog table, or purge_s3_path for a raw S3 prefix. Both methods automatically handle partition cleanup.

```python
# Delete files older than 30 days for a specific partition predicate
glue_ctx.purge_table(
    database="ecommerce",
    table_name="orders_parquet",
    options={
        "retentionPeriod": 720,           # 30 days in hours
        "partitionPredicate": "order_date < '2023-12-01'",
        "excludeStorageClasses": ["GLACIER"],
        "manifestFilePath": "s3://my-bucket/manifests/purge/"
    },
    transformation_ctx="purge_old_orders"
)

# Purge a raw S3 path directly
glue_ctx.purge_s3_path(
    s3_path="s3://my-bucket/tmp/",
    options={"retentionPeriod": 24},  # delete files older than 1 day
    transformation_ctx="purge_tmp"
)
```

--------------------------------

### UnnestFrame.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Flattens all nested struct fields to the top level and prepares join keys for array elements.

```APIDOC
## UnnestFrame.apply

### Description
Flattens all nested struct fields within a `DynamicFrame` to the top level, using dot-notation for field names. It also generates join keys for array elements to facilitate subsequent relational operations.

### Parameters
- **frame** (DynamicFrame) - The input DynamicFrame with nested structures.
- **transformation_ctx** (string) - An identifier for the transformation.

### Request Example
```python
from awsglue.transforms import UnnestFrame

# Input: {order_id: 1, address: {street: "123 Main", city: "NY"}}
# Output: {order_id: 1, address.street: "123 Main", address.city: "NY"}
flat_dyf = UnnestFrame.apply(
    frame=nested_dyf,
    transformation_ctx="unnest_frame"
)
flat_dyf.printSchema()
# root
# |-- order_id: long
# |-- address.street: string
# |-- address.city: string
```
```

--------------------------------

### GlueContext.purge_table / purge_s3_path

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Remove old data from S3. `purge_table` targets AWS Glue catalog tables, while `purge_s3_path` directly purges files from a specified S3 prefix.

```APIDOC
## GlueContext.purge_table / purge_s3_path

`purge_table(database, table_name, options)` deletes S3 files for a catalog table older than a retention period, automatically removing empty partitions from the catalog. `purge_s3_path` targets a raw S3 prefix.

```python
# Delete files older than 30 days for a specific partition predicate
glue_ctx.purge_table(
    database="ecommerce",
    table_name="orders_parquet",
    options={
        "retentionPeriod": 720,           # 30 days in hours
        "partitionPredicate": "order_date < '2023-12-01'",
        "excludeStorageClasses": ["GLACIER"],
        "manifestFilePath": "s3://my-bucket/manifests/purge/"
    },
    transformation_ctx="purge_old_orders"
)

# Purge a raw S3 path directly
glue_ctx.purge_s3_path(
    s3_path="s3://my-bucket/tmp/",
    options={"retentionPeriod": 24},  # delete files older than 1 day
    transformation_ctx="purge_tmp"
)
```
```

--------------------------------

### Apply Transformation Function with Map

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use `Map.apply` to transform each record in a DynamicFrame. The function should return the modified record or `None`/raise an exception to mark it as an error.

```python
from awsglue.transforms import Map

def enrich_record(record):
    # Normalize email to lowercase
    if "email" in record:
        record["email"] = record["email"].lower().strip()
    # Compute derived field
    record["full_name"] = f"{record.get('first_name', '')} {record.get('last_name', '')}".strip()
    # Remove PII
    if "ssn" in record:
        del record["ssn"]
    return record

enriched_dyf = Map.apply(
    frame=customers_dyf,
    f=enrich_record,
    transformation_ctx="enriched_dyf"
)
```

--------------------------------

### DynamicFrame.fromDF / DynamicFrame.toDF

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Converts between Spark DataFrames and AWS Glue DynamicFrames. `fromDF` converts a Spark DataFrame to a DynamicFrame, while `toDF` converts a DynamicFrame back to a Spark DataFrame, with options for resolving ambiguous types.

```APIDOC
## DynamicFrame.fromDF / DynamicFrame.toDF

`DynamicFrame.fromDF(dataframe, glue_ctx, name)` converts a Spark `DataFrame` to a `DynamicFrame`; `.toDF(options)` does the reverse, accepting `ResolveOption` specs to handle ambiguous choice types.

```python
from awsglue.dynamicframe import DynamicFrame, ResolveOption
from pyspark.sql.types import DoubleType

# DataFrame → DynamicFrame
spark_df = spark.read.parquet("s3://my-bucket/data/")
dyf = DynamicFrame.fromDF(spark_df, glue_ctx, "my_frame")

# Apply Spark SQL transformation, then convert back
filtered_df = dyf.toDF().filter("amount > 0")
clean_dyf = DynamicFrame.fromDF(filtered_df, glue_ctx, "clean_frame")

# toDF with choice resolution
resolved_df = dyf.toDF([
    ResolveOption("price", "Project", DoubleType()),
    ResolveOption("metadata", "KeepAsStruct")
])
```
```

--------------------------------

### Relationalize.apply

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Normalizes a nested DynamicFrame into relational tables by unnesting structs and pivoting arrays.

```APIDOC
## Relationalize.apply

### Description
Fully normalizes a nested or hierarchical `DynamicFrame` into a collection of flat relational tables. It achieves this by unnesting struct fields and pivoting array columns. The output is a `DynamicFrameCollection` containing a root table and separate tables for each array column.

### Parameters
- **frame** (DynamicFrame) - The input nested DynamicFrame.
- **staging_path** (string) - The S3 path for temporary storage during the relationalization process.
- **name** (string) - A base name for the output tables.
- **transformation_ctx** (string) - An identifier for the transformation.

### Request Example
```python
from awsglue.transforms import Relationalize, SelectFromCollection

# Nested input: orders with embedded line items array
dfc = Relationalize.apply(
    frame=nested_orders_dyf,
    staging_path=args["TempDir"] + "/relationalize/",
    name="orders",
    transformation_ctx="relationalize"
)

print(dfc.keys())
# dict_keys(['orders', 'orders_items'])

# Root table: one row per order
root_dyf = SelectFromCollection.apply(dfc, "orders", "select_root")

# Child table: one row per line item, with `orders.items.id` join key
items_dyf = SelectFromCollection.apply(dfc, "orders_items", "select_items")

# Join root and child
joined_dyf = root_dyf.join(
    paths1=["id"],
    paths2=["orders.items.id"],
    frame2=items_dyf,
    transformation_ctx="join_items"
)
```
```

--------------------------------

### DynamicFrame.mergeDynamicFrame

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Upserts records from a staging frame into the base frame. Records with matching primary keys are replaced by the staging version; unmatched source records are retained as-is.

```APIDOC
## DynamicFrame.mergeDynamicFrame

### Description
Upserts records from a staging frame into the base frame. Records with matching primary keys are replaced by the staging version; unmatched source records are retained as-is.

### Method Signature
`mergeDynamicFrame(stage_dynamic_frame, primary_keys)`

### Parameters
- **stage_dynamic_frame** (DynamicFrame) - The staging DynamicFrame containing records to merge.
- **primary_keys** (list) - A list of column names that constitute the primary key for matching records.

### Example
```python
# CDC pattern: merge incremental changes into the base dataset
base_dyf  = glue_ctx.create_dynamic_frame.from_catalog("prod", "customers")
delta_dyf = glue_ctx.create_dynamic_frame.from_catalog("prod", "customers_delta")

merged_dyf = base_dyf.mergeDynamicFrame(
    stage_dynamic_frame=delta_dyf,
    primary_keys=["customer_id"],
    transformation_ctx="merge_customers"
)

glu_ctx.write_dynamic_frame.from_catalog(
    frame=merged_dyf,
    database="prod",
    table_name="customers",
    transformation_ctx="write_merged"
)
```
```

--------------------------------

### Resolve Choice Transformation - Match Catalog

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Resolves 'ChoiceType' columns by matching the schema of a specified catalog table. Ensures type consistency with existing data.

```python
# Strategy 3: match schema from catalog table
resolved_dyf3 = ResolveChoice.apply(
    frame=raw_dyf,
    choice="match_catalog",
    database="ecommerce",
    table_name="orders_parquet",
    transformation_ctx="resolved_dyf3"
)
```

--------------------------------

### GlueContext.add_ingestion_time_columns

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Enhance DataFrames with ingestion time partition columns. This function appends columns like `ingest_year`, `ingest_month`, etc., to aid in organizing streaming data by arrival time.

```APIDOC
## GlueContext.add_ingestion_time_columns

`add_ingestion_time_columns(frame, time_granularity)` appends ingestion time partition columns (`ingest_year`, `ingest_month`, `ingest_day`, `ingest_hour`, `ingest_minute`) to a Spark `DataFrame`, useful for organizing streaming output by arrival time.

```python
from pyspark.sql import functions as F

# Add time columns up to minute granularity
enriched_df = glue_ctx.add_ingestion_time_columns(raw_df, "minute")

# enriched_df now has: ingest_year, ingest_month, ingest_day, ingest_hour, ingest_minute
enriched_dyf = DynamicFrame.fromDF(enriched_df, glue_ctx, "enriched")

glue_ctx.write_dynamic_frame.from_options(
    frame=enriched_dyf,
    connection_type="s3",
    connection_options={
        "path": "s3://my-bucket/events/",
        "partitionKeys": ["ingest_year", "ingest_month", "ingest_day"]
    },
    format="parquet"
)
```
```

--------------------------------

### Merge DynamicFrame with Primary Keys

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Upsert records from a staging DynamicFrame into a base DynamicFrame using primary keys. Matching records are updated, and unmatched records from the base are retained.

```python
# CDC pattern: merge incremental changes into the base dataset
base_dyf  = glue_ctx.create_dynamic_frame.from_catalog("prod", "customers")
delta_dyf = glue_ctx.create_dynamic_frame.from_catalog("prod", "customers_delta")

merged_dyf = base_dyf.mergeDynamicFrame(
    stage_dynamic_frame=delta_dyf,
    primary_keys=["customer_id"],
    transformation_ctx="merge_customers"
)

glue_ctx.write_dynamic_frame.from_catalog(
    frame=merged_dyf,
    database="prod",
    table_name="customers",
    transformation_ctx="write_merged"
)
```

--------------------------------

### Split Rows Based on Comparison

Source: https://context7.com/awslabs/aws-glue-libs/llms.txt

Use SplitRows to partition records into two DynamicFrames based on column-value comparisons. Records matching all predicates go to the first output ('name1'), and the rest go to the second ('name2').

```python
from awsglue.transforms import SplitRows, SelectFromCollection

# Separate high-value orders (amount > 1000) from the rest
dfc = SplitRows.apply(
    frame=orders_dyf,
    comparison_dict={"total_amount": ">" : 1000},
    name1="high_value",
    name2="standard",
    transformation_ctx="split_rows"
)

high_value_dyf = SelectFromCollection.apply(dfc, "high_value")
standard_dyf   = SelectFromCollection.apply(dfc, "standard")
```

--------------------------------

### Apply a Transform

Source: https://github.com/awslabs/aws-glue-libs/blob/master/awsglue/README.md

Transforms can be applied to DynamicFrames using the TransformClass.apply(args...) syntax. Ensure the TransformClass and its arguments are correctly specified.

```python
TransformClass.apply(args...)
```