Scenario C: Industrial IoT telemetry¶

Part of Streaming Transformations › Scenarios.

Goal¶

Industrial devices and gateways often send several readings in one message, and label each measurement with a raw tag ID. Analysts want one typed row per reading, labeled with the engineering name of the measurement, seconds after it arrives.

Transformations used¶

1. Lateral Flatten on the stream: expand the array of readings in each message, before the dimension join (derived table).
2. Stream enrichment: map each raw tag ID to its engineering name and unit.
3. Stream filtering on dimension columns: keep only readings that matched a known tag.
4. Insert-triggered materialized views: maintain per-minute rollups per device next to the raw readings.

Source data¶

Stream¶

The example uses these 3 records:

{"device_id": "press-01", "ts": "2026-10-09 08:00:00", "readings": [{"tag": "T100", "v": 71.4}, {"tag": "T200", "v": 12.8}, {"tag": "T300", "v": 0.42}]}
{"device_id": "press-02", "ts": "2026-10-09 08:00:00", "readings": [{"tag": "T100", "v": 68.9}, {"tag": "T200", "v": 11.1}, {"tag": "T999", "v": 3.0}]}
{"device_id": "press-01", "ts": "2026-10-09 08:00:30", "readings": [{"tag": "T100", "v": 72.0}, {"tag": "T200", "v": 13.2}]}

Dimension table¶

The PIPE joins this dimension table, refreshed periodically:

CREATE OR REPLACE TABLE dim_tags (
  tag_id STRING PRIMARY KEY RELY,
  tag_name STRING,
  unit STRING);

INSERT INTO dim_tags VALUES
  ('T100', 'spindle_temperature', 'degC'),
  ('T200', 'motor_current', 'A'),
  ('T300', 'vibration_rms', 'mm/s');

Target table¶

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

CREATE OR REPLACE TABLE iot_readings (
  device_id STRING, event_ts TIMESTAMP_NTZ, tag_id STRING,
  tag_name STRING, unit STRING, reading_value FLOAT);

Transformation¶

The PIPE synchronously lands data in a materialized view and the target table. Readings within device messages are flattened. Tag IDs are mapped to their engineering name and unit by joining the dimension table. Filtering on a dimension column keeps only readings that match a known tag.

-- Insert-triggered materialized view.
CREATE OR REPLACE MATERIALIZED VIEW iot_readings_per_minute
  REFRESH USING INSERT ONLY
  AS
  SELECT device_id,
         tag_name,
         DATE_TRUNC('minute', event_ts) AS minute_ts,
         AVG(reading_value) AS avg_value,
         COUNT(*) AS reading_count
  FROM iot_readings
  GROUP BY device_id, tag_name, DATE_TRUNC('minute', event_ts);

-- Pipe with flatten, enrichment and filter
CREATE OR REPLACE PIPE iot_readings_pipe AS
  COPY INTO iot_readings (device_id, event_ts, tag_id, tag_name, unit, reading_value) FROM (
    SELECT s.device_id, s.event_ts, s.tag_id, t.tag_name, t.unit, s.reading_value
    FROM (
      SELECT
        e.$1:device_id::STRING AS device_id,
        e.$1:ts::TIMESTAMP_NTZ AS event_ts,
        r.value:tag::STRING    AS tag_id,
        r.value:v::FLOAT       AS reading_value
      FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e,
           LATERAL FLATTEN(input => e.$1:readings) r) s
    LEFT OUTER JOIN dim_tags t
      ON s.tag_id = t.tag_id
    WHERE t.tag_id IS NOT NULL);

Result¶

Target table¶

Querying the target table returns these rows:

The T999 reading from press-02 is dropped because dim_tags has no matching tag. Each of the other seven readings lands as its own row with its tag name and unit.

Insert-triggered materialized views¶

The PIPE keeps iot_readings_per_minute up to date as it lands rows. The view holds one row per device, tag, and minute, with the average value and the number of readings. Querying it returns these rows:

The two press-01 records fall in the same minute, so their spindle temperature and motor current readings are combined into one row each, with the average value and a count of 2. As more readings arrive within the same minute, the average and the count are updated.

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. A FLATTEN before the dimension join must be wrapped in a derived table.
3. INNER JOIN isn’t supported. To keep only matched rows, use LEFT OUTER JOIN and WHERE on a dimension key with IS NOT NULL.
4. Each reading becomes its own row, so a message with ten readings writes ten rows. Take this into account when sizing the pipe.
5. ERROR_LOGGING currently can’t be enabled on the target table of a pipe that uses LATERAL FLATTEN.
6. Create the materialized views before the pipe. Rows ingested before a view exists aren’t backfilled into it.

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 B: Observability logs and OpenTelemetry · Next: Scenario D: Security events