Modern APIs and IoT edge gateways frequently stream telemetry logs as semi-structured JSON payloads containing deeply nested objects. Storing these raw JSON strings inside relational databases makes querying slow and makes data analysis difficult. A high-performance JSON-to-SQL Pipeline parses incoming streams, flattens key-value objects, and writes values into structured database tables.

By writing Python pipelines that use batch upserts, we can process JSON streams efficiently.

1. Flattening Nested JSON Objects

We write a helper function to flatten nested dictionaries recursively, producing clean flat records:

# data_flattener.py
def flatten_json(y):
    out = {}

    def flatten(x, name=''):
        if type(x) is dict:
            for a in x:
                flatten(x[a], name + a + '_')
        elif type(x) is list:
            i = 0
            for a in x:
                flatten(a, name + str(i) + '_')
                i += 1
        else:
            # Strip trailing underscore
            out[name[:-1]] = x

    flatten(y)
    return out

2. Implementing Database Batch Upserts

The pipeline processes batches of raw events, flattens each record, and executes bulk upsert statements, minimizing write latency and database connection overhead.

Streaming Batch Inserts with PostgreSQL `COPY` & `execute_values`

When flattening deeply nested vendor JSON payloads into relational PostgreSQL tables, avoid row-by-row ORM inserts. Using psycopg2.extras.execute_values or asynchronous asyncpg.copy_records_to_table increases ingestion throughput by 25x–50x. Explore more in Freelance PostgreSQL & Supabase Programmer Guide.