Scenario B: Observability logs and OpenTelemetry¶

Part of Streaming Transformations › Scenarios.

Goal¶

Logs and OpenTelemetry data arrive in high volume, and much of it is debug noise. Attributes are nested in arrays, which makes them hard to query. Teams want to keep only the useful records, turn attributes into rows, attach the owning team, and keep a few views of the same stream current, for example errors only and counts per service.

Transformations used¶

1. Stream filtering on stream columns: drop debug-level and other noisy records.
2. Lateral Flatten on the stream: expand resource or span attributes into rows.
3. Stream enrichment: attach the owning team and service metadata.
4. Insert-triggered materialized views: keep several views on the same target table, for example errors only and counts per service and severity.

Source data¶

Stream¶

The example uses these four log records. Each record carries its resource attributes as an array:

{"timeUnixNano": 1791532800000000000, "severityNumber": 9, "severityText": "INFO", "resource": {"service.name": "checkout", "attributes": [{"key": "host.name", "value": "node-a"}, {"key": "k8s.pod", "value": "checkout-7f9c"}]}, "traceId": "4bf92f3577b34da6", "body": "cart loaded"}
{"timeUnixNano": 1791532801000000000, "severityNumber": 13, "severityText": "WARN", "resource": {"service.name": "checkout", "attributes": [{"key": "host.name", "value": "node-a"}, {"key": "k8s.pod", "value": "checkout-7f9c"}]}, "traceId": "4bf92f3577b34da6", "body": "payment retry 1"}
{"timeUnixNano": 1791532802000000000, "severityNumber": 5, "severityText": "DEBUG", "resource": {"service.name": "inventory", "attributes": [{"key": "host.name", "value": "node-b"}, {"key": "k8s.pod", "value": "inventory-5d2e"}]}, "traceId": "a3ce929d0e0e4736", "body": "cache hit"}
{"timeUnixNano": 1791532803000000000, "severityNumber": 17, "severityText": "ERROR", "resource": {"service.name": "inventory", "attributes": [{"key": "host.name", "value": "node-b"}, {"key": "k8s.pod", "value": "inventory-5d2e"}]}, "traceId": "a3ce929d0e0e4736", "body": "stock lookup timed out"}

Dimension table¶

The PIPE joins this dimension table, refreshed periodically:

-- Dimension table: owning team and tier per service.
CREATE OR REPLACE TABLE dim_services (
  service_name STRING PRIMARY KEY RELY,
  owning_team STRING,
  tier STRING);

INSERT INTO dim_services VALUES
  ('checkout', 'payments-team', 'tier-1'),
  ('inventory', 'supply-team', 'tier-2');

Target table¶

The PIPE lands transformed and filtered rows in this target table, with one row per record and attribute:

CREATE OR REPLACE TABLE otel_logs (
  event_ts TIMESTAMP_NTZ, service_name STRING, severity STRING,
  trace_id STRING, message STRING, attr_key STRING, attr_value STRING,
  owning_team STRING, tier STRING);

Transformation¶

The PIPE synchronously lands data in materialized views and the target table. Record attributes are flattened, and rows are enriched by joining the owning team and service metadata from the dimension table. Noisy records are dropped by filtering on a stream column.

-- Insert-triggered materialized views.
CREATE OR REPLACE MATERIALIZED VIEW otel_errors
  REFRESH USING INSERT ONLY
  AS
  SELECT event_ts, service_name, trace_id, message, attr_key, attr_value, owning_team
  FROM otel_logs
  WHERE severity = 'ERROR';

CREATE OR REPLACE MATERIALIZED VIEW otel_counts_by_service
  REFRESH USING INSERT ONLY
  AS
  SELECT service_name, severity, COUNT(*) AS record_count
  FROM otel_logs
  GROUP BY service_name, severity;

-- Pipe with flatten, enrichment and filtering
CREATE OR REPLACE PIPE otel_logs_pipe AS
  COPY INTO otel_logs (event_ts, service_name, severity, trace_id, message,
                       attr_key, attr_value, owning_team, tier) FROM (
    SELECT s.event_ts, s.service_name, s.severity, s.trace_id, s.message,
           s.attr_key, s.attr_value, d.owning_team, d.tier
    FROM (
      SELECT
        TO_TIMESTAMP_NTZ(e.$1:timeUnixNano::NUMBER, 9) AS event_ts,
        e.$1:resource:"service.name"::STRING          AS service_name,
        e.$1:severityText::STRING                     AS severity,
        e.$1:severityNumber::INT                      AS severity_number,
        e.$1:traceId::STRING                          AS trace_id,
        e.$1:body::STRING                             AS message,
        a.value:key::STRING                           AS attr_key,
        a.value:value::STRING                         AS attr_value
      FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e,
           LATERAL FLATTEN(input => e.$1:resource:attributes) a) s
    LEFT OUTER JOIN dim_services d
      ON s.service_name = d.service_name
    WHERE s.severity_number >= 13);

Result¶

Target table¶

Querying the target table returns rows like these:

The DEBUG and INFO records are dropped. Each remaining record lands as one row per attribute, with the owning team and tier of its service.

Insert-triggered materialized views¶

The PIPE keeps both insert-triggered materialized views up to date synchronously as it lands rows in the target table.

Querying otel_errors returns the records with severity ERROR:

Querying otel_counts_by_service returns the number of landed rows per service and severity:

The counts are per landed row, so a record with two attributes counts twice.

Considerations¶

1. Dimension data is a snapshot that refreshes periodically, so updates can take several minutes to appear. This works well for slow-moving reference data. Dynamic tables can be used as dimension tables.
2. Create the materialized views before the pipe. Rows ingested before a view exists aren’t backfilled into it.
3. ERROR_LOGGING currently can’t be enabled on the target table of a pipe that uses LATERAL FLATTEN.

See also¶

See Stream filtering, Stream enrichment, Lateral Flatten, and Insert-triggered materialized views for more information on the transformations used in this scenario.

Previous: Scenario A: AI agent observability · Next: Scenario C: Industrial IoT telemetry