JSONL for Analytics Pipelines

DuckDB - SQL Straight Against the File

DuckDB is an in-process analytical database: it runs inside your Python process or as a single CLI binary, with nothing to administer. read_ndjson treats a JSONL file as a table - no CREATE TABLE, no COPY, no staging step. read_json_auto does the same while also guessing whether the file is newline-delimited or a JSON array.

-- DuckDB CLI: no server, no import, no schema declaration
SELECT * FROM read_ndjson('data.jsonl') LIMIT 5;

-- A real aggregate over the raw file
SELECT
    json_extract_string(metadata, '$.country') AS country,
    count(*)    AS events,
    sum(amount) AS revenue
FROM read_ndjson('events/*.jsonl')
WHERE event_type = 'purchase'
GROUP BY 1
ORDER BY revenue DESC;

-- Let DuckDB detect the layout (array vs newline-delimited)
SELECT * FROM read_json_auto('unknown_shape.json');

-- Inspect what schema inference decided, then widen the sample if needed
DESCRIBE SELECT * FROM read_ndjson('events.jsonl');
SELECT * FROM read_ndjson('events.jsonl', sample_size = -1);

-- Skip malformed lines, and record which file each row came from
SELECT filename, count(*)
FROM read_ndjson('landing/*.jsonl', ignore_errors = true, filename = true)
GROUP BY 1;

Automatic schema inference is what makes this effortless and is also the thing to watch: DuckDB samples the file to decide types, so a column that is an integer for the first several thousand lines and a string afterwards gets inferred as BIGINT and then fails. Use sample_size = -1 on untrusted input, or declare columns explicitly for anything scheduled.

Querying Object Storage and Writing Parquet

The httpfs extension extends the same table functions to s3://, gs://, and HTTPS URLs, so a bucket of JSONL is queryable without downloading it. Paired with COPY ... TO ... (FORMAT PARQUET), DuckDB is also the shortest path from a raw landing zone to a columnar table.

-- Once per database: enable object storage access
INSTALL httpfs;
LOAD httpfs;
CREATE SECRET s3_landing (TYPE s3, PROVIDER credential_chain, REGION 'us-east-1');

-- Query a whole prefix of JSONL without downloading it
SELECT event_type, count(*) AS n
FROM read_ndjson('s3://<bucket>/landing/events/2026/07/*.jsonl')
GROUP BY 1;

-- Convert raw JSONL to partitioned Parquet in one statement
COPY (
    SELECT
        cast(user_id AS BIGINT) AS user_id,
        event_type,
        cast(ts AS TIMESTAMP)   AS event_ts,
        cast(ts AS DATE)        AS event_date,
        amount
    FROM read_ndjson('s3://<bucket>/landing/events/*.jsonl', union_by_name = true)
)
TO 's3://<bucket>/curated/events'
(FORMAT PARQUET, COMPRESSION ZSTD, PARTITION_BY (event_date), OVERWRITE_OR_IGNORE);

Queries against that Parquet copy skip whole files through partition pruning, skip row groups using column statistics, and read only the columns named in the SELECT. The JSONL original can do none of that, because a text file has no footer and no statistics. That is the whole argument for converting, and it costs one COPY statement.

Polars - Lazy Scans, Predicate and Projection Pushdown

pl.read_ndjson() is eager: it parses the entire file into memory, then filters. pl.scan_ndjson() is lazy: it returns a LazyFrame, which is a query plan rather than data, and nothing is read until collect(). By then Polars has seen the whole chain and can rewrite it. Projection pushdown means it never builds columns the query does not use; predicate pushdown means the filter runs as rows are read, so rejected rows never occupy memory. A lazy scan of a file far larger than RAM can answer a narrow question where the eager version fails outright.

import polars as pl

# EAGER - parses the whole file into memory first, then filters.
# Fine for a small file, fatal for a large one.
df = pl.read_ndjson("events.jsonl")
result = df.filter(pl.col("amount") > 100).select(["user_id", "amount"])

# LAZY - builds a plan; nothing is read until collect()
result = (
    pl.scan_ndjson("landing/events_*.jsonl")      # globs are one logical table
    .filter(pl.col("event_type") == "purchase")   # predicate pushdown
    .select(["user_id", "amount"])                # projection pushdown
    .group_by("user_id")
    .agg(pl.col("amount").sum().alias("revenue"))
    .collect()
)

# .explain() prints the optimized plan, showing what was pushed into the scan
# Stream JSONL straight to Parquet without ever holding it in memory
(pl.scan_ndjson("landing/events_*.jsonl")
   .with_columns(pl.col("ts").str.to_datetime().alias("event_ts"))
   .sink_parquet("curated/events.parquet", compression="zstd"))

Be honest about the limits on a text format: JSONL has no footer, so Polars still walks every byte to find line boundaries. Pushdown saves materialization, not I/O - often the difference between a job that finishes and one killed by the out-of-memory reaper, but not the skipping a columnar format can do.

When Not to Query JSONL Directly

In-place querying is right when the data is read once or twice, or when the point is to avoid standing up infrastructure. It is wrong the moment the same file becomes something people query every day.

• Dashboards. Every refresh re-parses every byte; convert once and pay the parsing cost once.
• Point lookups. No index, no sort order. One row by id still means reading the file.
• Concurrent readers. Ten analysts scanning the same prefix pay the scan ten times, and a cloud warehouse bills each one.

Count readers. One reader, once: query in place. Many readers, repeatedly: convert. The format comparison covers where JSONL sits against Parquet, Avro, and CSV.

Loading JSONL into BigQuery

# Python: Load JSONL to BigQuery
from google.cloud import bigquery

client = bigquery.Client()
table_id = "project.dataset.users"

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
    autodetect=True,  # Auto-detect schema
    write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
)

# Load from local file
with open("users.jsonl", "rb") as source_file:
    job = client.load_table_from_file(source_file, table_id, job_config=job_config)

job.result()  # Wait for completion
print(f"Loaded {job.output_rows} rows")

# Load from Google Cloud Storage
uri = "gs://my-bucket/data/*.jsonl"
job = client.load_table_from_uri(uri, table_id, job_config=job_config)
job.result()

# Streaming inserts (real-time)
rows_to_insert = [
    {"id": 1, "name": "Alice", "email": "[email protected]"},
    {"id": 2, "name": "Bob", "email": "[email protected]"}
]

errors = client.insert_rows_json(table_id, rows_to_insert)
if errors:
    print(f"Errors: {errors}")

# Query nested JSON
query = """
SELECT
    user_id,
    JSON_VALUE(metadata, '$.country') as country,
    JSON_VALUE(metadata, '$.city') as city
FROM `project.dataset.events`
WHERE DATE(timestamp) = CURRENT_DATE()
"""

results = client.query(query)
for row in results:
    print(f"{row.user_id}: {row.country}, {row.city}")

Choosing the load path. Per Google's BigQuery pricing documentation, batch loading into BigQuery storage is free while the streaming insert API is billed per volume ingested. That settles most designs: write JSONL to a bucket and run a load job, and reserve streaming inserts for genuinely real-time cases. An external table skips loading entirely, at the cost of a full parse on every query.

Gotchas. autodetect infers the schema from a sample rather than the whole file, so a field appearing only in later records can be missing from the table entirely - declare the schema explicitly for anything scheduled. Point WRITE_TRUNCATE at a partition rather than a whole table, or a retry quietly wipes history. And prefer many medium files over one enormous one, because BigQuery parallelizes the load across files.

Schema Design for JSON Data

Promote anything you filter, join, or group on into a real column, and leave open-ended payloads in a native JSON column so a new key needs no migration. Then partition on event date, because on-demand pricing bills bytes scanned and an unpartitioned table makes every query pay for the whole history.

-- Create table with schema
CREATE OR REPLACE TABLE `project.dataset.users` (
    id INT64,
    email STRING,
    name STRING,
    metadata JSON,  -- Native JSON type
    tags ARRAY<STRING>,
    address STRUCT<
        street STRING,
        city STRING,
        country STRING
    >,
    created_at TIMESTAMP
)
PARTITION BY DATE(created_at)
CLUSTER BY email;

-- Query JSON fields
SELECT
    id,
    JSON_EXTRACT_SCALAR(metadata, '$.signup_source') as signup_source,
    JSON_EXTRACT_ARRAY(metadata, '$.interests') as interests
FROM `project.dataset.users`
WHERE JSON_EXTRACT_SCALAR(metadata, '$.plan') = 'premium';

-- Flatten nested arrays
SELECT
    id,
    tag
FROM `project.dataset.users`,
UNNEST(tags) as tag;

ETL DAG with JSONL

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.transfers.local_to_gcs import LocalFilesystemToGCSOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'etl_to_bigquery',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    start_date=datetime(2026, 1, 1),
    catchup=False
)

def extract_data(**context):
    """Extract data to JSONL"""
    import json
    import psycopg2

    conn = psycopg2.connect("postgresql://...")
    cursor = conn.cursor()

    execution_date = context['execution_date']
    filepath = f"/tmp/data_{execution_date.strftime('%Y%m%d')}.jsonl"

    cursor.execute("SELECT * FROM users WHERE updated_at::DATE = %s", (execution_date.date(),))

    with open(filepath, 'w') as f:
        for row in cursor:
            record = {'id': row[0], 'email': row[1], 'name': row[2]}
            f.write(json.dumps(record) + '\n')

    return filepath

def transform_data(**context):
    """Transform JSONL data"""
    import json

    filepath = context['task_instance'].xcom_pull(task_ids='extract')
    output_path = filepath.replace('.jsonl', '_transformed.jsonl')

    with open(filepath, 'r') as infile, open(output_path, 'w') as outfile:
        for line in infile:
            record = json.loads(line)
            record['email'] = record['email'].lower()
            record['processed_at'] = datetime.utcnow().isoformat()
            outfile.write(json.dumps(record) + '\n')

    return output_path

extract = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag
)

transform = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag
)

upload_to_gcs = LocalFilesystemToGCSOperator(
    task_id='upload_to_gcs',
    src="{{ task_instance.xcom_pull(task_ids='transform') }}",
    dst="data/{{ ds }}/users.jsonl",
    bucket='my-data-bucket',
    dag=dag
)

load_to_bq = BigQueryInsertJobOperator(
    task_id='load_to_bigquery',
    configuration={
        "load": {
            "sourceUris": ["gs://my-data-bucket/data/{{ ds }}/users.jsonl"],
            "destinationTable": {
                "projectId": "my-project",
                "datasetId": "analytics",
                "tableId": "users"
            },
            "sourceFormat": "NEWLINE_DELIMITED_JSON",
            "writeDisposition": "WRITE_APPEND",
            "autodetect": True
        }
    },
    dag=dag
)

extract >> transform >> upload_to_gcs >> load_to_bq

Gotchas. Writing intermediates to /tmp works only while every task runs on the same machine, which stops being true the moment you move to Celery or Kubernetes executors - push the file to object storage and pass the URI. Keep XCom for small values like that URI, because it is backed by the metadata database.

Dagster and Prefect - Asset-Centric Alternatives

Airflow is not the only option. Dagster is asset-oriented: you declare the data assets that should exist, such as a raw JSONL partition and the curated table derived from it, and Dagster works out execution order and tracks lineage. Prefect stays closer to ordinary Python, adding retries and observability to functions you already wrote.

# Dagster: declare the assets, not the task graph
from dagster import asset, AssetExecutionContext, DailyPartitionsDefinition
import polars as pl

daily = DailyPartitionsDefinition(start_date="2026-01-01")

@asset(partitions_def=daily)
def raw_events(context: AssetExecutionContext) -> str:
    day = context.partition_key
    return extract_to_jsonl(day, f"s3://<bucket>/landing/events/dt={day}/events.jsonl")

@asset(partitions_def=daily)
def curated_events(raw_events: str) -> None:
    pl.scan_ndjson(raw_events).sink_parquet(raw_events + ".parquet")

The payoff is backfills: when the unit of work is a declared asset partition, rerunning one bad day is a first-class operation and the lineage graph names what needs rebuilding.

Processing JSONL with PySpark

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.appName("JSONLProcessor").getOrCreate()

# Read JSONL
df = spark.read.json("s3://bucket/data/*.jsonl")

# Schema inference
df.printSchema()

# Transformations
transformed_df = df \
    .withColumn("created_date", to_date(col("created_at"))) \
    .withColumn("email_domain", split(col("email"), "@").getItem(1)) \
    .filter(col("status") == "active")

# Nested JSON operations
df_exploded = df \
    .select(
        col("id"),
        explode(col("tags")).alias("tag")
    )

# Aggregations
summary = df \
    .groupBy("email_domain") \
    .agg(
        count("*").alias("user_count"),
        avg("age").alias("avg_age")
    ) \
    .orderBy(desc("user_count"))

# Write to JSONL (partitioned)
transformed_df.write \
    .mode("overwrite") \
    .partitionBy("created_date") \
    .json("s3://bucket/output/")

# Write to Parquet for efficiency
transformed_df.write \
    .mode("overwrite") \
    .partitionBy("created_date") \
    .parquet("s3://bucket/parquet/")

Declare the schema in production. Inference costs an extra pass over the data before the real job starts, and it is nondeterministic: if a field is absent from every record in today's batch it is absent from today's schema, and a downstream job that selects it fails.

from pyspark.sql.types import StructType, StructField, StringType, LongType

schema = StructType([
    StructField("event_id",   StringType(), False),
    StructField("user_id",    LongType(),   True),
    StructField("event_type", StringType(), True),
    StructField("_corrupt_record", StringType(), True),
])

df = (spark.read
        .schema(schema)                   # no inference pass, deterministic
        .option("mode", "PERMISSIVE")     # keep bad lines instead of failing
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .json("s3://bucket/landing/events/*.jsonl"))

# Quarantine the failures, then continue with the clean rows
df.filter(df._corrupt_record.isNotNull()).write.mode("append").text("s3://bucket/quarantine/")
clean = df.filter(df._corrupt_record.isNull()).drop("_corrupt_record")

When not to use Spark at all. A cluster earns its overhead when the data genuinely exceeds one machine or the job needs a distributed shuffle. Below that, DuckDB or Polars on one large instance finishes the same transformation sooner and cheaper. Reach for Spark because the data demands it, not because it is nominally called big.

Data Quality

• Validate JSON syntax before loading
• Implement schema validation
• Handle nulls and missing fields
• Deduplicate records
• Add data quality checks in pipelines

Performance

• Partition data by date for faster queries
• Compress JSONL files (gzip, snappy)
• Use columnar formats (Parquet) for analytics
• Implement incremental loads
• Cluster tables on frequently queried columns

Reliability

• Implement idempotent pipelines
• Add retry logic with exponential backoff
• Monitor pipeline failures
• Maintain data lineage
• Version your data and schemas

Cost Optimization

• Use data lifecycle policies
• Archive old data to cold storage
• Optimize query patterns
• Monitor storage costs
• Use appropriate compression