How to Flatten OpenTelemetry Metrics in Spark 4x faster

Replace multiples explode_outer calls with a single generate operator

TL;DR

Nested JSON never arrives flat. Open telemetry metrics are the textbook case: one payload carries an array of resources, each resource carries scopes, each scope carries metrics, each metric carries data points. On Spark, you land the whole thing in a single VARIANT column and unroll it later. That unroll is what most raw-to-refined jobs actually spend their time doing.

The obvious way is one explode_outer per level. It works. It is also the part of the job that quietly costs you the most. The rest of this is the mechanical rewrite from N nested explode_outer calls into a single explode_outer of a flatten/transform tree, plus the one subtlety that makes the naive version silently wrong. General Spark technique for nested arrays in VARIANT or JSON.

The working example below uses a public OpenTelemetry payload.

Why chaining explode_outer calls hurts cost and correctness

Cost. Each explode_outer compiles to a Generate operator. Four nested arrays mean four Generate hops, with parent columns — including the raw payload — carried between them.

Correctness. explode drops rows for empty arrays; explode_outer preserves the parent and emits a null child for null or empty input. Any rewrite must reproduce that behavior exactly.

Mapping the nested array structure of an OTLP payload

payload (VARIANT column `raw`)
└── $.resourceMetrics[]                          rm
    └── $.scopeMetrics[]                         sm
        └── $.metrics[]                          metric
            └── $.sum|gauge|summary|histogram|
                exponentialHistogram .dataPoints[]   dp   <- leaf

Chain form:  raw --Generate--> rm --Generate--> sm --Generate--> metric --Generate--> dp
               (4 Generate hops, fat parent row carried through each)
Single form: raw --transform/flatten (no Generate)--> array<struct<rm,sm,metric,dp>>
               --Generate--> leaf
               (1 Generate hop, narrow leaf struct)

Before: chaining four explode_outer calls

Assume a table otlp_metrics_raw(payload_id STRING, raw VARIANT) with one payload per row:

from pyspark.sql import functions as F
df = (
    spark.table("otlp_metrics_raw")
    .withColumn("rm", F.explode_outer(
        F.variant_get(F.col("raw"), "$.resourceMetrics", "array<variant>")))
    .withColumn("sm", F.explode_outer(
        F.variant_get(F.col("rm"), "$.scopeMetrics", "array<variant>")))
    .withColumn("metric", F.explode_outer(
        F.variant_get(F.col("sm"), "$.metrics", "array<variant>")))
    .withColumn("dp", F.explode_outer(F.coalesce(
        F.variant_get(F.col("metric"), "$.sum.dataPoints", "array<variant>"),
        F.variant_get(F.col("metric"), "$.gauge.dataPoints", "array<variant>"),
        F.variant_get(F.col("metric"), "$.summary.dataPoints", "array<variant>"),
        F.variant_get(F.col("metric"), "$.histogram.dataPoints", "array<variant>"),
        F.variant_get(F.col("metric"), "$.exponentialHistogram.dataPoints", "array<variant>"),
    )))
)

The rewrite rule: build the leaf struct from the inside out

1. At each parent level, transform the child array into an array of leaf structs, then flatten the result back to one level.

2. The leaf transform builds the struct — struct(rm AS rm, sm AS sm, m AS metric, dp AS dp) — carrying every ancestor handle the projection reads after the explode (need $.schemaUrl off the resource? Then rm must be in the struct).

3. The last step — and only the last step — is one explode_outer of that array.

4. Restore outer semantics on every array — the step people skip, and the whole point of this post.

The hidden risk: flatten and transform lack outer semantics

Swapping a chain of explode_outer calls for one explode_outer over a flatten(transform(...)) tree trades an operator with outer behavior for two without it. explode_outer on a null or empty array keeps the parent row and emits one NULL-child row; flatten and transform make no such promise:

• Empty inner array → the parent disappears. transform of an empty array yields an empty array, flatten drops it, and the parent never reaches the explode — explode_outer would have emitted one row.

• NULL inner array → the whole row collapses. flatten returns NULL if any element is NULL [4]. One resource missing scopeMetrics nulls the entire leaf array and takes every sibling leaf with it. The damage isn’t local to the bad element.

def null_to_singleton(array_expr: str) -> str:
    """explode_outer parity: NULL/empty child array -> one NULL placeholder."""
    return (
        f"CASE WHEN ({array_expr}) IS NULL OR size({array_expr}) = 0 "
        f"THEN array(CAST(NULL AS VARIANT)) ELSE ({array_expr}) END"
    )

Wrap every array in the chain — leaf included — and the rewrite is row-for-row identical to the chain it replaces.

After: one explode_outer call replaces the chain

METRIC_KINDS = ("sum", "gauge", "summary", "histogram", "exponentialHistogram")

def null_to_singleton(array_expr: str) -> str:
    return (
        f"CASE WHEN ({array_expr}) IS NULL OR size({array_expr}) = 0 "
        f"THEN array(CAST(NULL AS VARIANT)) ELSE ({array_expr}) END"
    )

def dp_array(metric: str) -> str:
    kinds = ",\n            ".join(
        f"variant_get({metric}, '$.{k}.dataPoints', 'array<variant>')" for k in METRIC_KINDS
    )
    return f"coalesce(\n            {kinds}\n          )"

LEAF_ARRAY_EXPR = """explode_outer(
    flatten(transform({rm_arr}, rm ->
      flatten(transform({sm_arr}, sm ->
        flatten(transform({m_arr}, m ->
          transform({dp_arr}, dp -> struct(rm AS rm, sm AS sm, m AS metric, dp AS dp))
        ))
      ))
    ))
  )""".format(
    rm_arr=null_to_singleton("variant_get(raw, '$.resourceMetrics', 'array<variant>')"),
    sm_arr=null_to_singleton("variant_get(rm, '$.scopeMetrics', 'array<variant>')"),
    m_arr=null_to_singleton("variant_get(sm, '$.metrics', 'array<variant>')"),
    dp_arr=null_to_singleton(dp_array("m")),
)

FLATTEN_SQL = """
SELECT
  payload_id,
  rm,
  sm,
  metric,
  dp,
  variant_get(metric, '$.name', 'string')  AS metric_name,
  variant_get(metric, '$.unit', 'string')  AS metric_unit,
  CAST(variant_get(dp, '$.startTimeUnixNano', 'long') / 1e9 AS TIMESTAMP)
                                           AS start_timestamp,
  CAST(variant_get(dp, '$.timeUnixNano', 'long') / 1e9 AS TIMESTAMP)
                                           AS event_timestamp,
  variant_get(dp, '$.asInt', 'long')       AS value_as_int,
  variant_get(dp, '$.asDouble', 'double')  AS value_as_double
FROM (
  SELECT
    payload_id,
    leaf.rm     AS rm,
    leaf.sm     AS sm,
    leaf.metric AS metric,
    leaf.dp     AS dp
  FROM (
    SELECT payload_id, {leaf_array_expr} AS leaf
    FROM otlp_metrics_raw
  )
)
""".format(leaf_array_expr=LEAF_ARRAY_EXPR)

flat = spark.sql(FLATTEN_SQL)

rm, sm, metric, and dp remain VARIANT handles in the projection; resource attributes, scope name, bucket counts, exemplars — anything you need later is a variant_get away, without touching the unroll.

General checklist for flattening any nested JSON array

1. List the arrays from root to leaf.

2. Define a leaf struct containing the leaf and every required ancestor.

3. Replace each null or empty array with a typed singleton-NULL array.

4. Build nested transform and flatten expressions from the inside out.

5. Call explode_outer exactly once, at the top.

6. Verify Generate counts and compare rows against the original chain.

Rule: N nested explode_outer calls become one explode_outer over a guarded flatten(transform(...)) tree that pre-builds the leaf structs.

FAQ

Q: How to check spark behavior before executing ?

A: Run explain() on both forms and count the Generate nodes. Four becomes one — the entire structural claim, verifiable in one command:

spark.sql(FLATTEN_SQL).explain()   # count `Generate` nodes: expect exactly 1
r.

Q: Does this technique require the VARIANT type?

A: No. transform, flatten, and explode_outer are standard Spark SQL functions. For parsed structs, replace variant_get with field access and cast each NULL placeholder to the array’s element type.

Q: Does the same rule apply to OTLP logs?

A: Yes. OTLP logs use three array levels: resourceLogs[] → scopeLogs[] → logRecords[] [3]. Remove one flatten(transform(...)) level and the corresponding leaf-struct field..

Q: Is one explode_outer always faster than a chain?

A: Almost always. It removes Generate operators but may widen the row before the remaining Generate and adds generated-SQL complexity. Benchmark both implementations on representative data.

Q: What happens if one array is left unguarded?

A: You can lose rows silently. An empty array removes its parent leaf, while a NULL inner array can null the entire flattened array and remove otherwise-valid sibling rows.

---

References

[1] opentelemetry-proto/examples/metrics.json — Apache-2.0. Raw: https://raw.githubusercontent.com/open-telemetry/opentelemetry-proto/main/examples/metrics.json

[2] opentelemetry/proto/metrics/v1/metrics.proto

[3] opentelemetry-proto/examples/logs.json

Spark SQL references

[4] Spark SQL, Built-in Functions — flatten — “Returns NULL if any element of the input array is NULL.”

[5] Spark SQL, Built-in Functions — transform — higher-order function; returns an empty array for empty input.

[6] Spark SQL, Built-in Functions — explode_outer — “Returns null for entries that are null or empty.”