Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Real-time data streaming expertise covering Kafka architecture, stream processing with Flink and Kafka Streams, exactly-once semantics, windowing strategies (tumbling, sliding, session), watermarks, late data handling, schema registry, consumer group management, and production deployment patterns.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
486 lines · 1,768 words
Real-time streaming is the architecture of processing data as it arrives, rather than in batches. This skill covers the design, implementation, and operation of streaming systems that handle millions of events per second with low latency, high reliability, and exactly-once processing guarantees.
Real-time data streaming expertise covering Kafka architecture, stream processing with Flink and Kafka Streams, exactly-once semantics, windowing strategies (tumbling, sliding, session), watermarks, late data handling, schema registry, consumer group management, and production deployment patterns.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the Streaming Architect skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
What it covers
Overview
Kafka Architecture
Exactly-Once Semantics
Windowing Strategies
Watermarks and Late Data
Schema Registry
Consumer Groups
Stream Processing Topology Design
Capacity Planning
Decision Framework
When to Use
Output Format
Example
Edge Cases
Producers -> Brokers (Partitioned Topics) -> Consumers (Consumer Groups)
Topic: "orders" (6 partitions, replication factor 3)
Partition 0: [msg0, msg6, msg12, ...] -> Leader: Broker 1, Followers: 2, 3
Partition 1: [msg1, msg7, msg13, ...] -> Leader: Broker 2, Followers: 1, 3
Partition 2: [msg2, msg8, msg14, ...] -> Leader: Broker 3, Followers: 1, 2
Partition 3: [msg3, msg9, msg15, ...] -> Leader: Broker 1, Followers: 2, 3
Partition 4: [msg4, msg10, msg16, ...] -> Leader: Broker 2, Followers: 1, 3
Partition 5: [msg5, msg11, msg17, ...] -> Leader: Broker 3, Followers: 1, 2from confluent_kafka.admin import AdminClient, NewTopic
admin = AdminClient({'bootstrap.servers': 'broker1:9092,broker2:9092,broker3:9092'})
# Partition count guidelines:
# - Target throughput / per-partition throughput = min partitions
# - More partitions = more parallelism but more overhead
# - Cannot reduce partitions after creation
# - Rule of thumb: start with 6-12 for moderate throughput, 30-100 for high
topics = [
NewTopic(
topic='events.raw',
num_partitions=12,
# ... (condensed) ...
}
),
]
admin.create_topics(topics)from confluent_kafka import Producer, KafkaError
import json
import time
class ReliableProducer:
"""Production-grade Kafka producer with retry and delivery guarantees."""
def __init__(self, bootstrap_servers, topic):
self.topic = topic
self.producer = Producer({
'bootstrap.servers': bootstrap_servers,
'acks': 'all', # Wait for all ISR replicas
'enable.idempotence': True, # Exactly-once producer
'max.in.flight.requests.per.connection': 5, # With idempotence, safe up to 5
# ... (condensed) ...
def __enter__(self):
return self
def __exit__(self, *args):
self.flush()from confluent_kafka import Consumer, KafkaError, TopicPartition
import json
class ReliableConsumer:
"""Production-grade Kafka consumer with manual offset management."""
def __init__(self, bootstrap_servers, group_id, topics):
self.consumer = Consumer({
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest',
'enable.auto.commit': False, # Manual offset commits
'max.poll.interval.ms': 300000, # 5 min max processing time
'session.timeout.ms': 45000,
# ... (condensed) ...
"""Commit current offsets (call after successful processing)."""
self.consumer.commit(asynchronous=False)
def close(self):
self.consumer.close()from confluent_kafka import Producer
# Transactional producer: atomic writes across multiple partitions/topics
producer = Producer({
'bootstrap.servers': 'broker1:9092',
'transactional.id': 'my-app-instance-1', # Unique per instance
'acks': 'all',
'enable.idempotence': True,
})
producer.init_transactions()
try:
producer.begin_transaction()
# ... (condensed) ...
producer.commit_transaction()
except Exception as e:
producer.abort_transaction()
raiseSource -> Kafka (idempotent producer)
-> Consumer (read_committed isolation)
-> Process
-> Transactional Producer (atomic write + offset commit)
-> SinkKey requirements:
Producer: enable.idempotence=true, transactional.id set
Consumer: isolation.level=read_committed
Processing: Deterministic transforms (same input = same output)
Offset management: Offsets committed within the same transaction as output
# Apache Flink windowing examples (conceptual, applicable to any framework)
# 1. Tumbling Window: Fixed-size, non-overlapping
# [0-5min] [5-10min] [10-15min] ...
# Use for: Periodic aggregation (counts per minute, hourly revenue)
events \
.key_by("user_id") \
.window(TumblingEventTimeWindows.of(Time.minutes(5))) \
.aggregate(CountAggregator())
# 2. Sliding Window: Fixed-size, overlapping by slide interval
# [0-10min] [5-15min] [10-20min] ... (10-min window, 5-min slide)
# Use for: Moving averages, smoothed metrics
events \
# ... (condensed) ...
events \
.key_by("user_id") \
.window(GlobalWindows.create()) \
.trigger(CountTrigger.of(100)) # Fire every 100 events
.aggregate(AccumulatingAggregator())// Tumbling window aggregation
KTable<Windowed<String>, Long> counts = events
.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.count(Materialized.as("event-counts"));
// Sliding window (hopping)
KTable<Windowed<String>, Double> avgValues = events
.groupByKey()
.windowedBy(
TimeWindows.ofSizeAndGrace(
Duration.ofMinutes(10), // window size
Duration.ofMinutes(2) // grace period for late data
).advanceBy(Duration.ofMinutes(1)) // slide/hop
# ... (condensed) ...
SessionInfo::new,
(key, event, session) -> session.addEvent(event),
(key, s1, s2) -> s1.merge(s2),
Materialized.as("user-sessions")
);Event Time: |-------|-------|-------|-------|
t=0 t=10 t=20 t=30
Processing Time progresses ->
Watermark = "I believe all events with event_time <= W have arrived"
With watermark delay of 5 seconds:
When processing-time clock is at T, watermark = max(event_times_seen) - 5s
Late data = event arriving after watermark has passed its event_time# Spark Structured Streaming: watermark + late data policy
from pyspark.sql.functions import window, col
windowed_counts = (
events
.withWatermark("event_time", "10 minutes") # Allow 10 min late
.groupBy(
window("event_time", "5 minutes"),
"event_type"
)
.count()
)
# Output modes with watermarks:
# ... (condensed) ...
query = windowed_counts.writeStream \
.outputMode("append") \
.format("parquet") \
.option("checkpointLocation", "/checkpoints/windowed") \
.start("/output/windowed_counts")| Strategy | Behavior | Use When |
|---|---|---|
| Drop late data | Discard events arriving after watermark | Metrics where approximation is acceptable |
| Side output | Route late data to separate stream | Need to reprocess late data differently |
| Allowed lateness | Keep window state beyond watermark | Need accurate counts, can tolerate delays |
| Retraction | Update previous results | Dashboard/materialized view updates |
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
# Schema Registry client
registry = SchemaRegistryClient({'url': '[reference URL]'})
# Define Avro schema
order_schema_str = """
{
"type": "record",
"name": "Order",
"namespace": "com.company.events",
"fields": [
# ... (condensed) ...
value=avro_serializer(
order_dict,
SerializationContext(topic, MessageField.VALUE)
),
)from confluent_kafka import Consumer
# Eager rebalancing (default): stop all consumers, reassign all partitions
# Cooperative rebalancing: incremental reassignment, minimal disruption
consumer = Consumer({
'bootstrap.servers': 'broker1:9092',
'group.id': 'order-processor',
'partition.assignment.strategy': 'cooperative-sticky',
# Options:
# 'range' - Assigns contiguous partitions per topic
# 'roundrobin' - Round-robin across all partitions
# 'cooperative-sticky' - Incremental rebalance, sticky assignment
})# Check consumer group lag
kafka-consumer-groups.shell-cmd --bootstrap-server broker1:9092 \
--group order-processor --describe
# Output:
# TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG HOST
# orders 0 15234 15240 6 /10.0.1.5
# orders 1 18902 19100 198 /10.0.1.6
# orders 2 12456 12456 0 /10.0.1.7
# Key metrics to monitor:
# - Consumer lag per partition (should be near 0 for real-time)
# - Consumer lag trend (growing = consumer cannot keep up)
# - Number of active consumers vs partitions
# - Rebalance frequency (too frequent = unstable consumers)Filter -> Map -> FlatMap -> Branch
Example: Event routing
Input: raw events
-> Filter: discard malformed
-> Map: parse JSON, extract fields
-> Branch:
purchase events -> purchase topic
click events -> clickstream topic
error events -> error topicAggregate -> Join -> Dedup
Example: Real-time dashboard
Input: page views
-> Window(5min tumbling)
-> GroupBy(page_url)
-> Count()
-> Output: page_view_counts per 5-min window
Example: Stream-stream join
Input A: orders
Input B: payments
-> Join on order_id within 1-hour window
-> Output: enriched order with payment status# Kubernetes deployment for Kafka Streams application
apiVersion: apps/v1
kind: Deployment
metadata:
name: order-processor
spec:
replicas: 6 # Match partition count or be a divisor
selector:
matchLabels:
app: order-processor
template:
spec:
containers:
- name: processor
# ... (condensed) ...
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0 # Zero-downtime deploymentProducer throughput per broker:
Network: min(network_bandwidth, disk_write_speed) / replication_factor
Consumer throughput:
Limited by: network_bandwidth, consumer processing speed, partition count
Example:
10 GB/s network per broker
3 brokers, replication factor 3
Producer throughput: 10 GB/s / 3 (replication) = ~3.3 GB/s per broker
Total cluster throughput: ~10 GB/s (limited by replication)
With 12 partitions, 6 consumers:
Each consumer handles 2 partitions
Per-consumer throughput needed: total_throughput / 6Storage per topic = throughput * retention_period * replication_factor
Example:
1 GB/hour throughput
7-day retention
Replication factor 3
Storage = 1 GB/h * 168h * 3 = 504 GB per topic
Total cluster storage = sum(all topics) + 20% overheadDo you need < 100ms latency? -> True streaming (Kafka + Flink/Kafka Streams)
Do you need exactly-once? -> Kafka transactions + Flink checkpoints
Is your logic pure SQL? -> Kafka Streams with ksqlDB
Do you need complex event processing? -> Flink CEP
Is it simple enrichment/routing? -> Kafka Streams or Kafka Connect + SMTs
Do you need to join streams? -> Flink (stream-stream joins are most robust)
Are you already in the Spark ecosystem? -> Spark Structured Streaming
Use this skill when:
Designing or implementing streaming architect solutions
Reviewing or improving existing streaming architect approaches
Making architectural or implementation decisions about streaming architect
Learning streaming architect patterns and best practices
Troubleshooting streaming architect-related issues
Do NOT use this skill when:
The question is about a fundamentally different technology domain
A more specific sibling skill covers the exact topic needed
The user needs a complete hands-on tutorial rather than expert guidance
# Streaming Architect Analysis
## Context Assessment
[Situation summary and constraints]
## Recommended Approach
[Primary recommendation with rationale]
## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]
## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]
## Next Steps
- [Immediate action item]
- [Follow-up action item]Input: "Help me implement streaming architect for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended streaming architect approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Legacy system integration: When streaming architect must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities
SKILL.md
486 lines · 1,768 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
Streaming Architect bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for