<jupyter_start><jupyter_text>JSON to Parquet and Reduce Number of FilesDISCLAIMER: This is sample code and is provided 'as is'. Should be used as a reference only.  1. Import the required libraries:<jupyter_code>import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame

glueContext = GlueContext(SparkContext.getOrCreate())
block_size = 128*1024*1024
page_size = 1024*1024<jupyter_output><empty_output><jupyter_text>2. Create dynamic frame from glue catalog table:Using push-down-predicates to read only the required source partitions. Job bookmarks could work here as well.<jupyter_code>rawdatasource = glueContext.create_dynamic_frame.from_catalog(
    database = "energy", 
    table_name = "raw_energy_json", 
    transformation_ctx = "rawdatasource",
    push_down_predicate = "(month == '7' and day in ('4','5','6'))",
    additional_options={
        'groupFiles':'inPartition',
        'groupSize':128*1024*1024
    })
rawdatasource.count()
rawdatasource.printSchema()<jupyter_output><empty_output><jupyter_text>3. CollateUsing repartition, if using scala coalesce would be a better choice. The right number of partitions depends on your data profile and desired output file size.<jupyter_code>df = rawdatasource.toDF().repartition(2)
repartitioned = DynamicFrame.fromDF(dataframe=df, glue_ctx=glueContext, name="repartitioned")<jupyter_output><empty_output><jupyter_text>4. Write to a curated bucket with parquet format. Using glueParquet vs Parquet as it is using optimized file item writer. Use either of this two options:- OPTION 1: Write data into s3. Handle updating partitions externally ( athena alter command or glue crawler):<jupyter_code>datasink = glueContext.write_dynamic_frame_from_options(
    frame = repartitioned, 
    connection_type = "s3", 
    connection_options = {
        "path": "s3://pablo.data.samples/energy/parquet_from_json/",
        "partitionKeys": ["month","day"]}, 
    format = "glueparquet",
    format_options = {
        "compression": "snappy", 
        "blockSize": block_size, 
        "pageSize" : page_size},
    transformation_ctx = "datasink")<jupyter_output><empty_output><jupyter_text>- OPTION 2: Write and create/update catalog table partitions. See for details and limitations: https://docs.aws.amazon.com/glue/latest/dg/update-from-job.html<jupyter_code>datasink = glueContext.getSink(connection_type="s3",
                               path="s3://pablo.data.samples/energy/curated_glueparquet/",
                               enableUpdateCatalog=True,
                               updateBehavior="UPDATE_IN_DATABASE",
                               partitionKeys=["month", "day"],
                               transformation_ctx = "datasink")
datasink.setFormat(format="glueparquet", 
                   format_options = {
                        "compression": "snappy", 
                        "blockSize": block_size, 
                        "pageSize" : page_size})
datasink.setCatalogInfo(catalogDatabase="energy", catalogTableName="energy_table_curated_parquet")
datasink.writeFrame(repartitioned)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09656652360515021, 'fraction_numerical': 0.014101778050275904, 'mean_word_length': 3.8197932053175774, 'pattern_counts': {'":': 6, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 1, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '372777', 'license': 'Apache-2.0', 'n_tokens_mistral': 1002, 'n_tokens_neox': 936, 'n_words': 223, 'path': 'aws-glue/handling_small_files_compress_n_collate..ipynb', 'repo_name': 'imsanjoykb/1xEgm', 'size': '3262'}