How to init/update a table and create transformed files in the same PySpark glue job

Question

This seems like a really basic thing but I feel frustrated that I have not been able to figure it out. When it comes to writing dynamic frames to files and to the glue data catalog there are three options I understand: getSink, write_dynamic_frame_from_options and write_dynamic_frame_from_catalog.

I am reading the table from create_dynamic_frame.from_catalog set up using a glue crawler and I have bookmarks and partitions.

When I use getSink that means on subsequent runs in the same partition I am seeing duplicate files. Initially I hoped adding transformation context to each transformation would alleviate this problem but it persists. It seems if I am to achieve what I want with this API I have to dedupe the data and the code to do something like this is very intimidating for me a non-programmer.

However when I try to use a combination of the other two methods that also does not seem to work the catalog writer fails if the table does not already exists unlike the previous method which is permissive and creates one if it does not exist and I am not able to solve my duplicate file problem even after trying a few permutations of things I can no longer recall now.

What does work for me now is two separate crawlers and one glue job that only writes files. I am surprised there is no "out of the box" solution for such a basic pattern but I feel I might be missing something

re:Post Agent · Answer

To initialize/update a table and create transformed files in the same PySpark Glue job without creating duplicate files, you have a few options:

**Option 1: Using getSink with enableUpdateCatalog**

This approach allows you to write data and update the catalog in one step:

```python
sink = glueContext.getSink(
connection_type="s3", 
path="s3://your-bucket/path/", 
enableUpdateCatalog=True,
updateBehavior="UPDATE_IN_DATABASE",
partitionKeys=["your_partition_key"]
)
sink.setFormat("parquet")  # or your preferred format
sink.setCatalogInfo(
catalogDatabase="your_database",
catalogTableName="your_table"
)
sink.writeFrame(your_dynamic_frame)
```

**Option 2: Using job bookmarks to prevent duplicate processing**

To prevent duplicate files when processing the same partition multiple times, properly configure job bookmarks:

```python
# Initialize job with bookmarks
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read with transformation context
datasource = glueContext.create_dynamic_frame.from_catalog(
database="your_database",
table_name="your_table",
transformation_ctx="unique_identifier"
)

# Your transformations here
transformed_frame = apply_transformations(datasource)

# Write with the same transformation context
glueContext.write_dynamic_frame.from_options(
frame=transformed_frame,
connection_type="s3",
connection_options={
"path": "s3://your-bucket/output-path/",
"partitionKeys": ["your_partition_key"]
},
format="parquet",
transformation_ctx="unique_identifier"
)

# Commit the job to save bookmark state
job.commit()
```

**Important considerations:**

1. When using job bookmarks, make sure to:
- Use DynamicFrame API (not DataFrame or Spark SQL) for reading data
- Include a unique transformation_ctx parameter for each ETL operation
- Don't change the transformation_ctx between job runs
- Set the job's maximum concurrent runs to 1

2. For the getSink approach, you can specify updateBehavior as "UPDATE_IN_DATABASE" to update an existing table or create a new one if it doesn't exist.

3. If you're still seeing duplicate files, ensure you're properly configuring the job bookmark option in the Glue job properties.

By following these patterns, you should be able to both update your catalog and write transformed files in a single job without creating duplicates.  
**Sources**  
[Updating the schema, and adding new partitions in the Data Catalog using AWS Glue ETL jobs - AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/update-from-job.html)  
[Troubleshoot AWS Glue job bookmarks | AWS re:Post](https://repost.aws/knowledge-center/glue-job-bookmarks)

How to init/update a table and create transformed files in the same PySpark glue job

Add your answer

Relevant content