aws--agent-toolkit-for-aws/plugins/aws-data-analytics/skills/ingesting-into-data-lake/references/format-specific-loading.md

Format-Specific Data Loading

Complete guide for reading and processing different file formats in Glue ETL jobs.

Overview

This reference covers format-specific configuration and code examples for loading data from various file formats into S3 Tables:

• CSV/TSV (delimited text files)
• JSON/JSONL (JavaScript Object Notation)
• Parquet (columnar format with embedded schema)
• Avro (row-based format with embedded schema)
• ORC (Optimized Row Columnar)

CSV and TSV Files

Basic CSV Reading

# CSV with custom delimiter
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("inferSchema", "true") \
    .load(args['source_path'])

TSV (Tab-Separated Values)

# TSV (tab-separated)
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "\t") \
    .load(args['source_path'])

CSV Options

Option	Value	Description
header	true/false	First row contains column names
delimiter	,, \t, |, etc.	Field separator character
inferSchema	true/false	Automatically detect column types
quote	" (default)	Character for quoting fields
escape	\ (default)	Escape character
nullValue	NULL, empty, etc.	String representing null values
dateFormat	yyyy-MM-dd	Date parsing format
timestampFormat	yyyy-MM-dd HH:mm:ss	Timestamp parsing format

Advanced CSV Example

# CSV with custom options
source_df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("quote", "\"") \
    .option("escape", "\\") \
    .option("nullValue", "NULL") \
    .option("dateFormat", "yyyy-MM-dd") \
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") \
    .option("mode", "DROPMALFORMED") \
    .load(args['source_path'])

JSON and JSONL Files

JSON Lines (JSONL)

One JSON object per line (most common):

# JSON Lines (one JSON object per line)
source_df = spark.read.format("json").load(args['source_path'])

Nested JSON Handling

Option A: Flatten Nested Structures

from pyspark.sql.functions import col

# Flatten nested JSON
flattened_df = source_df.select(
    col("customer.customer_id").alias("customer_id"),
    col("customer.name").alias("customer_name"),
    col("customer.email").alias("email"),
    col("order_id"),
    col("order_date"),
    col("amount")
)

Option B: Preserve as STRUCT

No transformation needed - Iceberg supports STRUCT types:

# Preserve nested structure (no transformation)
# Schema becomes:
# - order_id: BIGINT
# - customer: STRUCT<customer_id:BIGINT, name:STRING, email:STRING>
# - order_date: DATE
# - amount: DECIMAL

JSON Options

Option	Value	Description
multiLine	true/false	Parse multi-line JSON objects
mode	PERMISSIVE, DROPMALFORMED, FAILFAST	How to handle malformed records
dateFormat	yyyy-MM-dd	Date parsing format
timestampFormat	yyyy-MM-dd'T'HH:mm:ss.SSSXXX	Timestamp format

Array Handling

# Explode array into separate rows
from pyspark.sql.functions import explode

df_with_items = source_df.select(
    col("order_id"),
    explode(col("items")).alias("item")
).select(
    col("order_id"),
    col("item.product_id"),
    col("item.quantity"),
    col("item.price")
)

# Or preserve as ARRAY type in Iceberg
# Schema: items ARRAY<STRUCT<product_id:STRING, quantity:INT, price:DECIMAL>>

Parquet Files

Basic Parquet Reading

# Parquet (direct read, schema preserved)
source_df = spark.read.format("parquet").load(args['source_path'])

Partitioned Parquet

Spark automatically detects Hive-style partitions:

# Partitioned Parquet (Spark auto-detects partitions)
source_df = spark.read.format("parquet").load("s3://bucket/events/")
# Partitions like year=2024/month=01/ are automatically handled

Detect Partition Structure

For partitioned data with Hive-style partitioning (e.g., year=2024/month=01/day=15/):

Using Python regex:

import re

# Example S3 path: s3://bucket/events/year=2024/month=01/day=15/part-0000.parquet
sample_s3_path = "s3://bucket/events/year=2024/month=01/day=15/part-0000.parquet"

# Extract partition key-value pairs
path_pattern = r'(\w+)=([^/]+)'
partitions = re.findall(path_pattern, sample_s3_path)
# Result: [('year', '2024'), ('month', '01'), ('day', '15')]

partition_columns = [col for col, _ in partitions]
print(f"Detected partition columns: {partition_columns}")
# Output: ['year', 'month', 'day']

Using AWS CLI:

# List S3 paths to identify partition patterns
aws s3 ls s3://bucket/events/ --recursive | head -20

# Look for patterns like:
# year=2024/month=01/day=01/
# year=2024/month=01/day=02/

Partition Column Inference

• Partition columns should typically be: INT, STRING, or DATE types
• Common partition patterns: year, month, day, region, category
• Important: Partition columns will NOT appear in the data files themselves (they're in the path)

Present Partition Info to User

Detected partitioned data structure:
- Partition columns: year (INT), month (INT), day (INT)
- Data columns: event_id, event_type, timestamp, user_id, properties
- Sample partition: year=2024/month=01/day=15
- Estimated partitions: ~90 (covering 3 months)

Avro Files

Basic Avro Reading

# Avro format
source_df = spark.read.format("avro").load(args['source_path'])

Avro Schema Extraction

Avro files contain embedded schemas. Extract and display:

Using Python avro library:

import avro.datafile
import avro.io
import json

# Read Avro file and extract schema
with open('downloaded-sample.avro', 'rb') as f:
    reader = avro.datafile.DataFileReader(f, avro.io.DatumReader())
    schema_json = reader.meta.get('avro.schema').decode('utf-8')
    schema = json.loads(schema_json)

    print("Avro Schema:")
    print(json.dumps(schema, indent=2))

    # Extract field names and types
    for field in schema['fields']:
        print(f"  {field['name']}: {field['type']}")

Using fastavro:

import fastavro

with open('downloaded-sample.avro', 'rb') as f:
    reader = fastavro.reader(f)
    schema = reader.writer_schema
    for field in schema['fields']:
        print(f"  {field['name']}: {field['type']}")

Avro to Iceberg Type Mapping

Avro Type	Iceberg Type	Notes
int	INTEGER	32-bit signed integer
long	BIGINT	64-bit signed integer
float	FLOAT	32-bit floating point
double	DOUBLE	64-bit floating point
boolean	BOOLEAN	Direct mapping
string	STRING	Direct mapping
bytes	BINARY	Direct mapping
fixed	BINARY	Fixed-length byte array
enum	STRING	Store enum values as strings
array<T>	ARRAY<T>	Direct mapping with recursive type
map<string, T>	MAP<STRING, T>	Direct mapping
record	STRUCT	Nested structure
union [null, T]	Nullable T	Avro nullable pattern
union [T1, T2, ...]	STRING	Multiple types → JSON string

Handling Avro Union Types

Avro uses unions for nullable fields:

// Avro schema with nullable field
{
  "name": "age",
  "type": ["null", "int"]
}

Maps to Iceberg:

age INT  -- Nullable by default in Iceberg

For complex unions (non-nullable):

from pyspark.sql.functions import col, when

# Example: Handle union of int and string
df_with_union = source_df.withColumn(
    "age_clean",
    when(col("age").cast("int").isNotNull(), col("age").cast("int"))
    .otherwise(None)
)

Options for complex unions:

• Option A: Convert to JSON string and store as STRING
• Option B: Flatten union types into separate columns (age_int, age_string)
• Option C: Fail and ask user how to handle

Present Avro Schema to User

Detected Avro schema with 15 fields:
- user_id (long) → BIGINT
- username (string) → STRING
- age (union[null, int]) → INT (nullable)
- status (enum: active, inactive) → STRING
- metadata (map<string, string>) → MAP<STRING, STRING>
- preferences (record) → STRUCT

Glue Job Configuration for Avro

Option A: Use --datalake-formats (spark-avro built-in in Glue 5.1 or higher):

# In job DefaultArguments
'--datalake-formats': 'iceberg,delta,hudi,avro'

Option B: Provide spark-avro JAR:

# In create-job command
--default-arguments '{
  "--extra-jars": "s3://my-bucket/jars/spark-avro_2.12-3.4.0.jar"
}'

ORC Files

Basic ORC Reading

# ORC format
source_df = spark.read.format("orc").load(args['source_path'])

ORC files include embedded schema similar to Parquet. No special configuration needed.

Sampling Source Data

Before loading, sample source files to understand structure:

CSV Sampling

# Download and inspect first 10 lines
aws s3 cp s3://<bucket>/<key> - | head -10

Parquet Schema Inspection

import pyarrow.parquet as pq

# Read Parquet schema
table = pq.read_table('s3://<bucket>/<key>')
print(table.schema)

# Sample first 10 rows
df = table.to_pandas()
print(df.head(10))

JSON Sampling

# Download and inspect first 5 JSON objects
aws s3 cp s3://<bucket>/<key> - | head -5

Complete Glue ETL Script Template

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'source_path', 'target_table', 'source_format'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read source data based on format
if args['source_format'] == 'csv':
    source_df = spark.read.format("csv") \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .load(args['source_path'])
elif args['source_format'] == 'json':
    source_df = spark.read.format("json").load(args['source_path'])
elif args['source_format'] == 'parquet':
    source_df = spark.read.format("parquet").load(args['source_path'])
elif args['source_format'] == 'avro':
    source_df = spark.read.format("avro").load(args['source_path'])
elif args['source_format'] == 'orc':
    source_df = spark.read.format("orc").load(args['source_path'])
else:
    raise ValueError(f"Unsupported format: {args['source_format']}")

# Apply transformations as needed
transformed_df = source_df.select(
    # Column transformations here
)

# Write to Iceberg table
transformed_df.writeTo(args['target_table']).append()

job.commit()

Format-Specific Common Issues

CSV Issues

Issue: Column type inference incorrect Solution: Explicitly specify schema or cast columns after reading

Issue: Quoted fields not parsed correctly Solution: Set .option("quote", "\"") and .option("escape", "\\")

JSON Issues

Issue: Multi-line JSON not parsing Solution: Set .option("multiLine", "true")

Issue: Malformed JSON records Solution: Set .option("mode", "DROPMALFORMED") or "PERMISSIVE"

Parquet Issues

Issue: Partition columns not detected Solution: Verify path follows Hive-style partitioning (key=value/)

Issue: Schema evolution errors Solution: Use .option("mergeSchema", "true") when reading

Avro Issues

Issue: Avro library not found Solution: Add --datalake-formats: iceberg,avro to job arguments

Issue: Complex union types failing Solution: Convert to STRING or handle with conditional logic

Best Practices

1. Always sample data first: Understand structure before loading
2. Validate schema mapping: Ensure source types map correctly to Iceberg
3. Handle malformed records: Use appropriate error handling mode
4. Test with small dataset: Verify transformations work before full load
5. Monitor CloudWatch logs: Check for parsing errors or warnings
6. Document format-specific options: Keep track of delimiter, quote char, etc.
7. Use schema evolution carefully: Understand impact on existing data

Summary

Different file formats require different reading configurations:

Format	Key Considerations	Primary Options
CSV/TSV	Delimiter, header, quotes	delimiter, header, quote
JSON	Nested structures, arrays	multiLine, flatten vs preserve
Parquet	Partition detection	Auto-detected, mergeSchema
Avro	Union types, embedded schema	--datalake-formats: avro
ORC	Similar to Parquet	Auto-schema, minimal config

With format-specific configuration, Glue ETL can successfully load data from any supported format into S3 Tables.