<jupyter_start><jupyter_text>Working with unpredictibly changing schemas This notebook describe sample code snippets for handling source schema changes in AWS Glue. DISCLAIMER: This is sample code and is provided 'as is'. Should only be used as reference and we assume no responsability for errors<jupyter_code>import sys
from awsglue.transforms import *
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

glueContext = GlueContext(SparkContext.getOrCreate())<jupyter_output>Starting Spark application<jupyter_text>Source dataset could have new columns over time.One option is to treat new columns as custom fields. To keep output dataset consistant we can add them as one field with a self defining format (e.g. JSON). Keep in mind that new columns cannot grow indefinitely as there could be limitations on the size fo the JSON string/document.  1. Use Dynamicframe to read latest schema directly from the Glue Catalog. New columns will be updated via an AWS Glue Crawler. Assuming the crawler configuration allows updates when new columns are identified.<jupyter_code>sales_source = glueContext.create_dynamic_frame.from_catalog(database="sales", table_name="sales")
sales_source.printSchema()<jupyter_output><empty_output><jupyter_text>2. Use the Map function to rebuild schema and keep new fields within a struct object.<jupyter_code>import json

static_fields = ['date', 'categories','geo','revenue']
custom_fields = []
custom_fields_nm = 'custom_fields'

def build_struct(rec):
    rec[custom_fields_nm] = {}
    for key in rec:
        if key in static_fields:
            rec[key] = rec[key]
        elif key != custom_fields_nm:
            rec[custom_fields_nm][key] = json.dumps(rec[key])
            if key not in custom_fields:
                custom_fields.append(key)
                          
    #remove custom fields from the root
    for k in custom_fields:
        del rec[k]
    return rec

sales_with_struct = Map.apply(frame = sales_source, f = build_struct)
sales_with_struct.printSchema()

sales_with_struct.toDF().show(20,False)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.06669776119402986, 'fraction_numerical': 0.0018656716417910447, 'mean_word_length': 4.444162436548224, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '343124', 'license': 'Apache-2.0', 'n_tokens_mistral': 611, 'n_tokens_neox': 587, 'n_words': 222, 'path': 'aws-glue/working_with_changing_schemas.ipynb', 'repo_name': 'imsanjoykb/1xEgm', 'size': '2144'}