agentsclimarketplace

Fabric eventstream

Skill wardawgmalvicious/claude-config/skills/fabric-eventstream

Personal Claude Code config — skills, subagents, hooks, and rules for Microsoft Fabric and Power BI workflows on Windows. Cherry-pickable, no semver.

Install
npx -y skills add wardawgmalvicious/claude-config --skill fabric-eventstream

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

  • 2 stars2 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

What its author says it does

Copied from the file, not written here

Use for Microsoft Fabric Eventstream — the streaming-ingestion item routing CDC / Event Hubs / Kafka / IoT / HTTP / MQTT events into Lakehouse, Eventhouse, Activator, or derived streams, and how external apps produce events to a schema-associated custom endpoint. Covers source connectors (Azure SQL / SQL MI / PostgreSQL / MySQL / MongoDB / Cosmos DB CDC, Mirrored DB Delta CDF preview, Event Hubs / IoT Hub / Kafka / MSK / Confluent / Kinesis / Service Bus / MQTT / HTTP / Solace), DeltaFlow analytics-ready CDC (auto-table + schema evolution), Activator destination + in-Eventstream `Set Alert` flow, three workspace-monitoring KQL tables (`EventStreamNodeStatus`/`EventStreamMetrics`/`EventStreamErrorMetrics`, ~6h status lag) + republish-on-enable, mTLS Key Vault on Kafka, custom-endpoint CloudEvents producer format (binary mode — `cloudEvents:` app properties, `dataschema` version routing), and gotchas (republish required, 6h status lag, filter by ArtifactId not name, CloudEventPropertyMissingException).

SKILL.md

18.6 KB, as published. Nobody here has run it

Fabric Eventstream

Streaming-data ingestion item that pulls events from a wide source surface (CDC / Event Hubs / Kafka / IoT / HTTP / MQTT) and routes them into Fabric destinations (Lakehouse, Eventhouse, Activator, derived stream, custom endpoint). Authoring is graph-based: source nodes → optional transformations → destination nodes, edited then published to go live.

When to use vs not

Use Eventstream when the data is arriving as events and needs routing or transformation before it lands. Skip it when the data is bulk / batch (use a Data Pipeline Copy activity), already in the lake (use Spark / SQL directly), or when the only consumer is a Mirrored Database in append-only mode (mirroring lands data straight in OneLake without an Eventstream).

For real-time analytics on the resulting events, pair an Eventstream with fabric-eventhouse (KQL Database). For real-time rules, pair with an Activator destination (covered below).

Authoring model

  • Edit mode vs Live mode: changes only take effect after Publish. New nodes added in Edit mode produce no traffic until publish.
  • Sources = where events come from. Transformations = inline filter / aggregate / GroupBy / Manage Fields / SQL. Destinations = where events go. Each destination can have its own format (Delta / JSON / Avro) where applicable.
  • Permissions: workspace Contributor or higher to author; Viewer can read Data insights monitoring on a published stream.
  • Virtual-network injection (private-network sources): use Eventstream connector VNet injection for sources behind a firewall — see Microsoft Learn.

Source connectors

SourceNotes
Azure SQL DB CDCRequires sys.sp_cdc_enable_db; do NOT also enable Mirroring on the same DB
Azure SQL Managed Instance CDCSame shape as Azure SQL DB CDC
SQL Server on VM CDCPublic-net or VNet-injected
PostgreSQL CDCAzure DB for PostgreSQL, Amazon RDS / Aurora PostgreSQL, GCP Cloud SQL — logical replication required
MySQL DB CDCAzure DB for MySQL
MongoDB CDC (preview)Specify collections to monitor; initial snapshot + tail
Azure Cosmos DB CDCContainer-level change feed
Mirrored Database Delta CDF (preview, April 2026)New: stream row-level inserts/updates/deletes from a Mirrored Database's Delta Change Data Feed into Eventstream. Toggle via Mirrored DB config dashboard → Delta table managementEnable delta change data feed, or via enableDeltaChangeDataFeed in the Mirrored DB REST API. Connector reference: extended capabilities.
Azure Event Hubs / IoT HubNative sources — no CDC layer
Apache Kafka / Amazon MSK / Confluent Cloud KafkaKafka-protocol sources — base connector GA (June 2026); SASL_SSL / SASL_PLAINTEXT / Microsoft Entra auth. Custom-CA / mTLS is still preview — see below
Amazon Kinesis Data StreamsSingle-shard or multi-shard
Azure Service BusQueue or topic subscription — GA (June 2026)
Google Cloud Pub/Sub
Solace PubSub+
MQTT (preview)
HTTP (preview)Stream from external platforms via standard HTTP requests; predefined public feeds available
Real-time weatherFabric-hosted demo source
Azure Data ExplorerPull from an existing ADX table

DeltaFlow — analytics-ready CDC events (preview)

Available on Azure SQL CDC, Azure SQL MI CDC, SQL Server on VM CDC, and PostgreSQL CDC. When the schema-handling step is set to Analytics-ready events & auto-updated schema, DeltaFlow transforms raw Debezium CDC events into a tabular shape mirroring the source table, enriched with:

  • Op / change-type column: insert / update / delete
  • Event-timestamp column

Extras you get for free:

  • Automatic destination table management — when routing to a supported destination (e.g. an Eventhouse), tables are auto-created matching the source schema.
  • Schema evolution — new source columns and new tables propagate to registered schemas and destination tables without manual intervention.

Without DeltaFlow you receive raw Debezium envelopes and have to flatten them yourself.

Destinations

DestinationUse when
LakehouseLand events as Delta files for batch analytics
Eventhouse / KQL DatabaseReal-time KQL queries; pair with fabric-eventhouse
ActivatorRule-based alerts and automation (see below)
Derived streamChain a downstream Eventstream — useful for fan-out and reusable transforms
Custom endpointPush to an external Event Hubs / Kafka / AMQP-compatible system

Activator destination — set alert directly in Eventstream (preview)

Configure rules in-place without leaving Eventstream. Add an Activator destination, then select the alert icon on it to open the Rules pane:

  • View all rules linked to this Eventstream's Activator item
  • Stop / start a rule with the toggle
  • Edit / delete via the menu
  • Add rule at the bottom of the pane
  • Open in Activator to manage activation history and test notifications

Rule condition shapes:

Check valueWhen the action fires
on each eventEvery event flowing through the stream
On each event whenEvents matching a single-field condition (e.g. No_Empty_Docs == 0)
On each event grouped bySame condition, evaluated per group on a chosen field (e.g. Neighborhood)

Actions: Teams message, email, webhook, Power Automate, custom action.

Workspace monitoring (preview) — KQL tables

Enable workspace monitoring (Workspace settings → MonitoringLog workspace activity) and Fabric auto-creates a monitoring Eventhouse with three Eventstream-specific tables. Republish any Eventstream that existed before monitoring was enabled — pre-existing streams emit nothing until they're republished.

TableCadenceWhat it tells you
EventStreamNodeStatus~6 hoursEach node's running / paused / failed state
EventStreamMetrics1 minuteIncoming / outgoing message counts, bytes, watermark delay, backlog
EventStreamErrorMetrics1 minuteError counts by type (runtime, deserialization, conversion)

All three tables share base dimensions: Timestamp, ArtifactId, ArtifactName, WorkspaceId, WorkspaceName, CustomerTenantId, Level, OperationId, PremiumCapacityId, PlatformMonitoringCategory, PlatformMonitoringTableName, LogAnalyticsResourceId. Filter by ArtifactId / WorkspaceId — name columns can lag after rename / move.

// Most-recent status per node in one Eventstream
EventStreamNodeStatus
| where ArtifactId == "<eventstream-artifact-id>"
| summarize arg_max(Timestamp, *) by NodeId
| project Timestamp, NodeName, NodeDirection, NodeType, NodeStatus
| order by NodeDirection asc

// Incoming vs outgoing in 5-minute windows
EventStreamMetrics
| where ArtifactId == "<eventstream-artifact-id>"
| where MetricsName in ("Incoming Messages", "Outgoing Messages")
| summarize TotalMessages = sum(Value)
    by TimeWindow = bin(Timestamp, 5m), MetricsName
| order by TimeWindow asc

// Recent errors grouped by type
EventStreamErrorMetrics
| where ArtifactId == "<eventstream-artifact-id>"
| where Timestamp > ago(24h) and Value > 0
| summarize TotalErrors = sum(Value)
    by TimeWindow = bin(Timestamp, 5m), MetricsName, NodeDirection
| order by TimeWindow desc

For ad-hoc per-node visualizations during authoring, the Data insights tab on the lower pane of the Eventstream editor surfaces metrics directly — works without workspace monitoring enabled but is per-node and not historical.

Custom CA / mTLS for Kafka connectors (preview)

For Kafka, Amazon MSK, and Confluent Cloud Kafka sources, you can specify a custom CA certificate and a client certificate sourced from Azure Key Vault to enforce TLS / mTLS. Configured in the source connection step. Use when the broker is behind a private CA or requires client-cert auth.

Producing to a schema-associated custom endpoint (CloudEvents)

This is the producer side — how an external app must format events it pushes to a custom-endpoint source. The Eventstream authoring side (adding the source, wiring destinations) is above; this section is what the sending code has to get right. Applies only when the custom endpoint has an associated schema (a schema group / EventDefinition set). The portal's authoritative reference is the endpoint's Show sample code → Event Hub tab, which emits CloudNative.CloudEvents SDK code.

Verified end-to-end (2026-07-07) by pushing records and reading them back via Kusto; this wire format is not documented on Microsoft Learn (the extended-features docs describe the UI, not the format).

Binary content mode is required — not structured

The endpoint's Azure Stream Analytics EventHub input adapter reads CloudEvents attributes from the Event Hub message's application properties (CloudEvents AMQP binary content mode), not from the JSON body. Structured mode — the whole CloudEvent in the body with ContentType=application/cloudevents+json — is silently ignored: the adapter still hunts for a type property, doesn't find it, and drops the event with:

Microsoft.Streaming.AzureStreamAnalytics.Adapters.Input.EventHub.Exceptions.CloudEventPropertyMissingException: CloudEvent property type is missing.

Correct per-event shape (Azure.Messaging.EventHubs.EventData)

  • Body = the data payload JSON only (just the record fields — not a wrapped CloudEvent).
  • ContentType = application/json.
  • Application properties, each prefixed cloudEvents: (the CloudEvents AMQP binding convention):
PropertyValueNotes
cloudEvents:specversion1.0
cloudEvents:typeschema name, e.g. SLTermsSelects the schema — must exactly match a schema id in the associated set (case-sensitive)
cloudEvents:sourceany non-empty URIValue unconstrained by the schema envelope
cloudEvents:idfresh GUID per eventCloudEvents requires source+id unique
cloudEvents:dataschemahttps://<host>.messagingcatalog.azure.net/schemagroups/<EventDefinition artifactId>/schemas/<type>/versions/<vN>Required to route to a table — the /versions/vN segment supplies {CloudEventSchemaVersion}

The portal sample copies the attributes generically:

foreach (var attr in cloudEvent.GetPopulatedAttributes())
    eventData.Properties[$"cloudEvents:{attr.Key}"] = attr.Value?.ToString();

Where the schema-group base URI comes from

The https://<host>.messagingcatalog.azure.net/schemagroups/<id> base of dataschema is the only piece the producer can't derive — the code appends /schemas/{type}/versions/{vN} itself. Provenance:

  • It's the Fabric-auto-provisioned Azure Schema Registry ("messaging catalog") endpoint for the eventstream's schema group; the trailing GUID is the schema-group ID tied to the workspace's Event Schema Set item.
  • Not surfaced in the Fabric portal UI except inside the custom endpoint's Show sample code → Event Hub tab — copy it from there (verified 2026-07-08).
  • Not present in the git-synced Eventstream definition either: the .Eventstream folder (eventstream.json, eventstreamProperties.json, .platform) carries schemaMode and the {CloudEventType}_{CloudEventSchemaVersion} table template but no messagingcatalog host and no schema-group GUID (verified 2026-07-08). The CustomEndpoint source's properties is {}. So the sample code is currently the only confirmed source; whether the live REST eventstream_get_definition returns more than the git sync is untested.

Two independent gates

  1. Envelope gate — the adapter finds type in the application properties. Fails with CloudEventPropertyMissingException if attributes are in the body or not cloudEvents:-prefixed.
  2. Schema-validation gate — the body fields must match the Avro schema types. All-string schemas pass easily; non-string fields (Avro bytes / boolean) reject mismatched JSON values. A failure here shows a generic "dropped per schema registry error policy" diagnostic (not the envelope exception).

Destination table naming (Eventhouse)

A schema-associated eventstream → Eventhouse (processed ingestion) auto-creates one table per schema, named {CloudEventType}_{CloudEventSchemaVersion} — e.g. SLTerms_v1, SLProdcodes_v2. The version comes from the dataschema /versions/vN segment.

Version-bump gotcha

Editing a schema in the set mints a new version (it does not edit in place). The dataschema URI must point at the current version, and versions can differ across schemas in the same set (observed: SLTerms / SLCarriers at v1, SLProdcodes at v2 after a bytesstring edit). Point at the wrong version → the event validates against the old version's types → dropped. (Open question: whether Fabric accepts a latest form in dataschema to avoid pinning — untested.)

Reference C# implementation: sytebridge.core/Helpers/AzureEventHubPusher.cs (SendBatch sets the cloudEvents:* props; ExportToAzureEventHub is the SDK path) and sytebridge.core/Models/Job/JobOutput.cs (BuildDataSchema).

Gotchas

IssueCauseFix
Existing Eventstream emits no monitoring dataStream was published before workspace monitoring was enabledOpen in editor and Republish — required once per pre-existing stream
Monitoring tables don't appear after enablingDatabase refresh delayWorkspace settings → Monitoring → toggle off then on
ArtifactName / WorkspaceName show stale valuesName columns cached from emission timeFilter / join by ArtifactId / WorkspaceId only
EventStreamNodeStatus shows old status after a node failedStatus is emitted ~every 6 hoursFor real-time status, use the Eventstream editor's live view
CorrelationId maps to multiple nodesAdvanced processing (e.g. SQL operator with multiple destinations)Disambiguate using NodeDirection + NodeType together with CorrelationId
No detailed log messages in monitoringPreview limitation — only metrics + error countsUse the editor's runtime logs for the message text; full diagnostic logs are planned
Mirrored DB CDC source rejectedCan't enable Mirroring AND Eventstream CDC on same DBPick one — the docs explicitly call this out
New Activator rule doesn't fireEventstream wasn't republished after adding the destinationRepublish the Eventstream after wiring the destination
Connector behind firewall failsSource not publicly reachableUse Eventstream connector VNet injection
DeltaFlow not available on a CDC sourceCurrently scoped to Azure SQL / SQL MI / SQL Server VM / PostgreSQL CDCUse raw mode for other CDC sources and flatten Debezium yourself
CloudEventPropertyMissingException: ...type is missing when pushing to a schema-associated custom endpointAttributes sent in the body / structured modeUse CloudEvents binary mode: cloudEvents:-prefixed application properties (esp. cloudEvents:type), body = payload JSON only. See Producing to a schema-associated custom endpoint above
Event associates in Data preview but no table is writtenMissing / wrong cloudEvents:dataschemaSet dataschema to the current schema version URI (/versions/vN) — it's what routes to a table
Event dropped after editing a schemaThe schema edit bumped the versionUpdate dataschema /versions/vN to the new current version; versions can differ per schema in the same set

Reference

See also

  • fabric-eventhouse — the natural KQL-Database pair for analytics on streamed events
  • fabric-rest-api — Eventstream item REST endpoints, LRO polling, jobType values
  • fabric-auth — token audience for Fabric REST against Eventstream items
  • fabric-monitoring — Workspace monitoring, broader L1/L2 monitoring picture beyond Eventstreams

Keep looking

Skills are one crate of 328,083. Ordering is by how many stacks a row turns up in, so the top of any crate is what has actually been picked rather than what has the most stars.