OpenTelemetry Message Decoder

The OpenTelemetry message decoder converts OpenTelemetry JSON payloads into Pinot rows during realtime ingestion. It supports OpenTelemetry logs, metrics, and traces through a single decoder class selected by the otel.data.type decoder property.

ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder

The decoder expects OpenTelemetry JSON payloads shaped like OTLP JSON export messages. It does not run an OTLP collector endpoint by itself; use a stream such as Kafka or another supported realtime source to deliver the JSON payloads to Pinot.

When to Use

Use the OpenTelemetry decoder when:

Logs, metrics, or traces are already exported as JSON messages into a stream.
You want Pinot tables to support observability analytics without writing a custom transform pipeline.
You need OpenTelemetry attributes available as either JSON blobs or promoted Pinot columns.

Use the Prometheus decoder or Prometheus remote-write decoder for Prometheus-specific payloads.

Decoder Property

Set the decoder class and choose one data type:

{
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "logs"
}

Supported otel.data.type values:

Value	Input payload	Extractor
`logs`	`resourceLogs` payloads	OpenTelemetry log records.
`metrics`	`resourceMetrics` payloads	Gauge and sum metric points.
`traces`	`resourceSpans` payloads	Span records.

The value is case-insensitive in practice because it is normalized before matching.

Log Ingestion

OpenTelemetry log payloads are decoded from resourceLogs[].scopeLogs[].logRecords[]. Recommended schema:

{
  "schemaName": "otel_logs",
  "dimensionFieldSpecs": [
    {"name": "severityText", "dataType": "STRING"},
    {"name": "body", "dataType": "STRING"},
    {"name": "traceId", "dataType": "STRING"},
    {"name": "spanId", "dataType": "STRING"},
    {"name": "attributes", "dataType": "JSON"},
    {"name": "resourceAttributes", "dataType": "JSON"},
    {"name": "service.name", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "severityNumber", "dataType": "INT"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ts",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

Realtime table stream config:

{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-logs",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "logs",
  "stream.kafka.consumer.prop.auto.offset.reset": "smallest"
}

Log output fields:

Field	Description
`ts`	Log timestamp converted from `timeUnixNano`. If missing or unparseable, the decoder uses ingest time.
`severityNumber`	OpenTelemetry severity number.
`severityText`	OpenTelemetry severity text.
`body`	Log body value converted to string-compatible form.
`traceId`	Trace ID associated with the log record.
`spanId`	Span ID associated with the log record.
`attributes`	Remaining log attributes as JSON.
`resourceAttributes`	Resource attributes as JSON.
Additional columns	If a schema column matches an attribute key, the decoder can promote that attribute into the column.

Example query:

SELECT ts, severityText, body, "service.name"
FROM otel_logs
WHERE severityText = 'ERROR'
  AND "service.name" = 'checkout'
ORDER BY ts DESC
LIMIT 100

Metrics Ingestion

OpenTelemetry metric payloads are decoded from resourceMetrics[].scopeMetrics[].metrics[]. Recommended schema:

{
  "schemaName": "otel_metrics",
  "dimensionFieldSpecs": [
    {"name": "metric", "dataType": "STRING"},
    {"name": "labels", "dataType": "JSON"},
    {"name": "unit", "dataType": "STRING"},
    {"name": "description", "dataType": "STRING"},
    {"name": "service.name", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "value", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ts",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

Stream config:

{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-metrics",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "metrics"
}

Metric output fields:

Field	Description
`metric`	Metric name.
`labels`	Data point attributes as JSON.
`value`	Numeric gauge or sum data point value.
`unit`	Metric unit, when present.
`description`	Metric description, when present.
`ts`	Data point timestamp converted from `timeUnixNano`.

Example query:

SELECT DATETRUNC('minute', ts) AS minute, AVG(value)
FROM otel_metrics
WHERE metric = 'cpu.util'
  AND JSON_EXTRACT_SCALAR(labels, '$.host', 'STRING') = 'server-01'
GROUP BY minute
ORDER BY minute

Trace Ingestion

OpenTelemetry trace payloads are decoded from resourceSpans[].scopeSpans[].spans[]. Recommended schema:

{
  "schemaName": "otel_traces",
  "dimensionFieldSpecs": [
    {"name": "Name", "dataType": "STRING"},
    {"name": "Traceid", "dataType": "STRING"},
    {"name": "spanid", "dataType": "STRING"},
    {"name": "parent_tid", "dataType": "STRING"},
    {"name": "context", "dataType": "JSON"},
    {"name": "events", "dataType": "JSON"},
    {"name": "service.name", "dataType": "STRING"},
    {"name": "http.method", "dataType": "STRING"},
    {"name": "http.status_code", "dataType": "STRING"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "start_time",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    },
    {
      "name": "end_time",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

Stream config:

{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-traces",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "traces"
}

Trace output fields:

Field	Description
`Name`	Span name.
`Traceid`	Trace ID.
`spanid`	Span ID.
`parent_tid`	Parent span ID.
`start_time`	Span start time converted from `startTimeUnixNano`.
`end_time`	Span end time converted from `endTimeUnixNano`.
`context`	Span attributes/context as JSON.
`events`	Span events as JSON.
Additional columns	If a schema column matches a span or resource attribute key, the decoder can promote it into that column.

Example latency query:

SELECT
  "service.name",
  PERCENTILETDIGEST(
    CAST((ToEpochMillis(end_time) - ToEpochMillis(start_time)) AS DOUBLE),
    95
  ) AS p95_ms
FROM otel_traces
WHERE start_time >= ago('PT1H')
GROUP BY "service.name"
ORDER BY p95_ms DESC

Attribute Modeling

For logs and traces, attribute handling follows two patterns:

Keep the full attributes object in attributes, resourceAttributes, or context JSON columns for flexible exploration.
Promote frequently filtered attributes to top-level Pinot columns by adding matching column names to the schema.

Example promoted fields:

{
  "dimensionFieldSpecs": [
    {"name": "service.name", "dataType": "STRING"},
    {"name": "deployment.environment", "dataType": "STRING"},
    {"name": "host.name", "dataType": "STRING"}
  ]
}

Add JSON indexes when you frequently filter inside JSON attribute columns. Add inverted indexes to promoted low-cardinality attributes.

Troubleshooting

Symptom	Cause	Fix
Decoder fails during init	`otel.data.type` is missing or not one of `logs`, `metrics`, `traces`.	Set `stream.<type>.decoder.prop.otel.data.type`.
Rows have missing timestamps	OpenTelemetry timestamp field is missing or unparseable.	Confirm the payload uses OTLP JSON timestamp fields such as `timeUnixNano`.
Attribute filters are slow	Query filters inside a JSON blob without an index.	Promote hot attributes to top-level columns or add a JSON index.
Expected attribute column is null	The schema column name does not match the OpenTelemetry attribute key.	Match names exactly, including punctuation such as `service.name`.
Only one telemetry type is decoded	One table/decoder instance handles one `otel.data.type`.	Use separate topics/tables for logs, metrics, and traces, or split upstream.

Get Started

Ingestion

External Tables

Query Data

Manage Data

Observability

Visualize Data

Cluster Operations

Manage Security

Release Notes

Reference

OpenTelemetry Message Decoder

When to Use

Decoder Property

Log Ingestion

Metrics Ingestion

Trace Ingestion

Attribute Modeling

Troubleshooting

​When to Use

​Decoder Property

​Log Ingestion

​Metrics Ingestion

​Trace Ingestion

​Attribute Modeling

​Troubleshooting

When to Use

Decoder Property

Log Ingestion

Metrics Ingestion

Trace Ingestion

Attribute Modeling

Troubleshooting