Add Native Sparkplug Protobuf Decoding in EventStream SQL Operator

Product: Microsoft Fabric Real-Time Intelligence (RTI) - Eventstream

Submitted by: QCells

Reference: Azure Stream Analytics built-in Protobuf deserializer (GA November 2024)

Scenario: Decode Sparkplug MQTT telemetry and transform in Eventstream

Feature: Process Sparkplug protobuf messages with SQL transformations in Eventstream
  As a Data Engineer building industrial analytics with Fabric RTI
  I want to upload a Sparkplug .proto file and write SQL queries in Eventstream
  So that I can decode binary MQTT payloads and apply real-time transformations

  Scenario: Upload Sparkplug schema and write SQL transformation
    Given I have an MQTT source publishing Sparkplug B protobuf-encoded messages
    When I upload the Sparkplug .proto definition file to Eventstream input
    And I configure the deserializer as "Protobuf" with the Sparkplug schema
    Then Eventstream should deserialize binary payloads into structured records
    And I can write SQL queries to SELECT, filter, aggregate, and JOIN deserialized fields
    And decoded fields (timestamp, node_id, device_id, metrics) are queryable in SQL
    And transformed results flow to Eventhouse/Lakehouse destination

Use Case Examples

1. QCells - Real-Time Solar Farm Monitoring

Challenge: QCells' solar installations stream device metrics (inverter temps, power output, fault states) via MQTT in Sparkplug protobuf format. Current Eventstream MQTT connector receives binary payloads; engineers must manually handle protobuf deserialization or use external tools, adding complexity.

Solution: Upload Sparkplug .proto definition file to Eventstream input; configure Protobuf deserializer. Write SQL query to filter/aggregate deserialized metrics on real-time stream.

Impact:

• Reduces setup time from 2-4 weeks to <4 hours
• Eliminates custom integration code
• Enables sub-second alerting on equipment anomalies

Example Setup:

1. Upload Sparkplug .proto file to Eventstream MQTT input
2. Configure input deserializer:

   Format: ProtobufSchema file: sparkplug_b.protoMessage type: org.eclipse.sparkplug.protobuf.Payload

3. Write SQL transformation query:

   SELECT   timestamp,  node_id,  device_id,  metric_name,  metric_value,  CASE     WHEN metric_name = 'temperature_C' AND CAST(metric_value AS FLOAT) > 75       THEN 'ALERT_OVERHEAT'    WHEN metric_name = 'power_kW' AND CAST(metric_value AS FLOAT) < 50       THEN 'ALERT_LOW_PRODUCTION'    ELSE 'OK'  END as alert_status,  System.Timestamp() as event_timeFROM mqtt_sparkplug_inputWHERE node_id = 'solar-plant-1'

Output to Eventhouse:

timestamp, node_id, device_id, metric_name, metric_value, alert_status, event_time
2026-08-07T14:33:52Z, solar-plant-1, inverter-12, temperature_C, 62.3, OK, 2026-08-07T14:33:53Z
2026-08-07T14:33:52Z, solar-plant-1, inverter-12, power_kW, 245.7, OK, 2026-08-07T14:33:53Z
2026-08-07T14:33:57Z, solar-plant-1, inverter-13, temperature_C, 78.1, ALERT_OVERHEAT, 2026-08-07T14:33:58Z

→ Real-time dashboard updates; triggered alerts sent to operations team

2. Siemens Industrial Automation

Challenge: Siemens PLC devices send production line telemetry via Sparkplug MQTT. Operators need to join equipment state with production KPIs, perform rolling aggregations, and trigger maintenance alerts in <100ms.

Solution: Deserialize Sparkplug → apply SQL transformation with JOIN to reference table (equipment master in Lakehouse) → alert on threshold breach.

Example SQL Query:

SELECT 
  e.equipment_name,
  s.device_id,
  s.metric_name,
  CAST(s.metric_value AS FLOAT) as current_value,
  e.maintenance_threshold,
  CASE 
    WHEN CAST(s.metric_value AS FLOAT) > e.maintenance_threshold 
    THEN 'SCHEDULE_MAINTENANCE' 
    ELSE 'OK' 
  END as maintenance_action,
  AVG(CAST(s.metric_value AS FLOAT)) OVER (
    PARTITION BY s.device_id, s.metric_name 
    ORDER BY System.Timestamp() 
    ROWS BETWEEN 10 PRECEDING AND CURRENT ROW
  ) as rolling_avg
FROM mqtt_sparkplug_input s
LEFT OUTER JOIN equipment_master e ON s.device_id = e.equipment_id
WHERE s.metric_name IN ('vibration_hz', 'temperature_C')

Output: Eventhouse table equipment_alerts triggering automatic work orders

3. General Electric (GE Digital) - Predictive Maintenance

Challenge: GE ingests machine telemetry from thousands of industrial assets; 80% use standard Sparkplug, 20% use proprietary Sparkplug extensions with custom headers or enrichment metadata.

Solution:

• Standard Sparkplug (80%): Built-in Protobuf deserializer + standard SQL
• Proprietary (20%): Custom C# deserializer that unwraps proprietary headers → deserializes Sparkplug core → enriches with site metadata

Impact: Single unified pipeline handles all asset types without forking architecture

Acceptance Criteria

AC1: Built-in Sparkplug Protobuf Deserializer

• [ ] Eventstream MQTT input supports Protobuf deserialization format selection
• [ ] Users can upload Sparkplug .proto file (or reference Sparkplug schema library)
• [ ] Deserializer converts binary Sparkplug payloads → JSON-compatible structured records
• [ ] Deserialized output includes all Sparkplug fields: timestamp, node_id, device_id, metrics[] (with name, value, type)
• [ ] Type mapping automatic: Sparkplug UINT32 → BIGINT, FLOAT → DOUBLE, STRING → VARCHAR, BOOLEAN → BIT
• [ ] Failed deserialization: configurable action (drop message, log error, route to dead-letter)
• [ ] Documentation: step-by-step guide for uploading Sparkplug schema + example queries

AC2: SQL Transformation on Deserialized Data

• [ ] Deserialized columns available directly in SQL SELECT without manual casting
• [ ] SQL supports filtering: WHERE device_id = 'inverter-12' AND metric_name = 'temperature_C'
• [ ] SQL supports aggregation with windowing: AVG(metric_value) OVER (PARTITION BY device_id ROWS BETWEEN X PRECEDING AND CURRENT ROW)
• [ ] SQL supports JOINS: LEFT OUTER JOIN reference_table ON deserialized_column = ref_column
• [ ] SQL supports complex logic: CASE statements, string functions, mathematical expressions on deserialized fields
• [ ] Temporal windowing: TUMBLE, HOP, SESSION for time-series aggregations

AC3: Configuration & Monitoring

• [ ] Input configuration UI: dropdown for "Serialization format" with "Protobuf" option
• [ ] After selecting Protobuf: file upload for .proto schema definition
• [ ] Configuration validation: test deserializer against sample MQTT message before saving
• [ ] Eventstream metrics: deserialization success/failure counts, latency (p50, p95, p99)
• [ ] Activity logs: schema upload events, deserialization failures with message context

AC4: Documentation & Examples

• [ ] Tutorial: "Decode Sparkplug MQTT data in Eventstream in 5 minutes"
• [ ] Sample SQL queries for common IoT patterns (threshold alerting, rolling aggregation, JOIN with reference data)
• [ ] Troubleshooting guide: common deserialization errors and solutions
• [ ] Example: QCells solar monitoring, Siemens PLC automation use cases with working SQL

Priority: P1 (High)

Business Value Justification

Market Context:

• Sparkplug adoption growing 40%+ YoY in industrial IoT (utilities, solar, wind, manufacturing, automotive)
• IEC 62541 MQTT standard; de-facto industrial IoT telemetry format
• Azure Stream Analytics proved Protobuf support (GA November 2024); Eventstream should inherit
• Fabric RTI lacks Sparkplug documentation/examples; perceived as unsupported by industrial customers
• Industrial IoT analytics market: $25B projected by 2027

Revenue & Strategic Impact:

• Retention: Prevents churn of industrial customers (QCells, utilities, renewable energy) to AWS/Databricks
• New Logo Revenue: Enables Fabric RTI to win industrial verticals standardizing on Sparkplug (estimated 20-30 enterprise customers in pipeline)
• Expansion: Existing RTI customers in discrete manufacturing can expand to Industrial analytics
• Customer Ease: Sparkplug .proto file upload + SQL examples drastically reduce time-to-value

Customer Economics:

• Eliminates 2-4 week custom integration/development work
• Reduces operational complexity: native deserializer vs. external tools
• Improves time-to-insight: MQTT → deserialize → SQL transform → Eventhouse in <5 seconds end-to-end
• SQL is familiar to data engineers; no custom code needed for standard use cases

Engineering ROI:

• Build effort: ~1-2 weeks (reuse Azure Stream Analytics Protobuf deserializer + Eventstream integration)
• Reuse: Existing protobuf deserializer from Azure Stream Analytics; Sparkplug schema from Eclipse Foundation
• Support burden: minimal (leverages standard Sparkplug spec; custom deserializers are customer responsibility)
• Addressable install base: 50+ active Fabric RTI industrial customers + 20-30 new logos in pipeline

Priority Justification Table

Factor Rating Rationale

Overall Priority: P1 

References & Resources

• Azure Stream Analytics Protobuf (GA November 2024): Generally Available: Protocol Buffers (Protobuf) with Azure Stream Analytics
• Protobuf Deserialization Documentation: Parse Protobuf - Azure Stream Analytics
• Custom Deserializer Example: Azure Stream Analytics Custom Deserializers GitHub
• MQTT Sparkplug Specification: MQTT Sparkplug
• Eclipse Sparkplug Repository: Eclipse Sparkplug GitHub (includes .proto definitions)
• Fabric Eventstream SQL Operator: Process Events Using a SQL Operator
• Fabric Eventstream Overview: Microsoft Fabric Eventstreams Overview