JSON Ingestion

JSON (JavaScript Object Notation) is a highly flexible, semi-structured data format widely used for APIs, event streaming, and nested datasets. Unlike CSV, JSON natively supports complex data structures (like maps, arrays, and nested structures) and carries type information implicitly.

Single-Line vs. Multi-Line JSON

Spark's default JSON reader is optimized to read Single-Line JSON (where each line of the file is a complete, independent JSON record).

• Single-Line JSON (Standard):

{"id":1, "name":"Alice", "skills":["PySpark","SQL"]}
{"id":2, "name":"Bob", "skills":["Java"]}

*Fast and parallelizable. Spark can split the file across multiple partitions easily.*

• Multi-Line JSON (Formatted/Pretty-printed):

[
  {
    "id": 1,
    "name": "Alice",
    "skills": ["PySpark", "SQL"]
  }
]

*Requires loading the entire file into a single executor thread. Slower and cannot be easily split, requiring the `multiLine` option set to `"true"`.*

PySpark Code Example: Reading Nested JSON

Here is a complete script demonstrating how to ingest nested JSON files, enforce schemas, and extract values from nested arrays:

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, ArrayType
from pyspark.sql.functions import col, explode

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("JSON Ingestion") \
    .master("local[*]") \
    .getOrCreate()

# 2. Define schema with nested arrays
json_schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("name", StringType(), True),
    StructField("skills", ArrayType(StringType()), True)
])

# 3. Read single-line JSON dataset
df = spark.read \
    .format("json") \
    .schema(json_schema) \
    .load("dataset.json")

df.show()
df.printSchema()

# 4. Explode array: Convert the array of skills into separate rows
exploded_df = df.select("name", explode("skills").alias("skill"))
exploded_df.show()

# 5. Write data back as compressed JSON
df.write \
    .format("json") \
    .mode("overwrite") \
    .option("compression", "bzip2") \
    .save("output_json_directory")