Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Hands-on streaming data engineering covering Apache Kafka producer/consumer tuning, Flink stateful stream processing, Spark Structured Streaming, windowing strategies (tumbling, sliding, session, global), exactly-once semantics, backpressure management, dead-letter queues, schema evolution, and production monitoring 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
496 lines · 1,761 words
Streaming data engineering is the discipline of building systems that process unbounded data in near-real-time. Unlike streaming architecture (which focuses on system design), this skill focuses on the hands-on engineering: writing producers and consumers, implementing windowed aggregations, guaranteeing exactly-once delivery, handling backpressure, and operating streaming jobs in production.
Hands-on streaming data engineering covering Apache Kafka producer/consumer tuning, Flink stateful stream processing, Spark Structured Streaming, windowing strategies (tumbling, sliding, session, global), exactly-once semantics, backpressure management, dead-letter queues, schema evolution, and production monitoring 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 Data Engineer 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 Producer Engineering
Consumer Engineering
Apache Flink Stream Processing
Spark Structured Streaming
Backpressure Management
Schema Evolution
Production Monitoring
Delivery Guarantees Decision Guide
Common Anti-Patterns
When to Use
Output Format
Example
Edge Cases
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
// Throughput tuning
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536); // 64KB batches
props.put(ProducerConfig.LINGER_MS_CONFIG, 20); // Wait up to 20ms to fill batch
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); // LZ4 for speed, zstd for ratio
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB buffer
// Reliability
props.put(ProducerConfig.ACKS_CONFIG, "all"); // All ISR replicas must ACK
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // Prevent duplicates
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); // Safe with idempotence
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
KafkaProducer<String, Event> producer = new KafkaProducer<>(props);// Custom partitioner for hot-key distribution
public class WeightedPartitioner implements Partitioner {
@Supersede
public int partition(String topic, Object key, byte[] keyBytes,
Object value, byte[] valueBytes, Cluster cluster) {
int numPartitions = cluster.partitionCountForTopic(topic);
if (key == null) return ThreadLocalRandom.current().nextInt(numPartitions);
String keyStr = (String) key;
// Spread hot keys across multiple partitions with suffix
if (isHotKey(keyStr)) {
int subPartition = ThreadLocalRandom.current().nextInt(4);
return Math.abs((keyStr + "-" + subPartition).hashCode()) % numPartitions;
}
return Math.abs(keyStr.hashCode()) % numPartitions;
}
}from confluent_kafka import Consumer, KafkaError, TopicPartition
consumer = Consumer({
'bootstrap.servers': 'broker1:9092',
'group.id': 'order-processor-v2',
'auto.offset.reset': 'earliest',
'enable.auto.commit': False, # Manual commit for exactly-once
'max.poll.interval.ms': 300000, # 5 min max processing time
'session.timeout.ms': 45000,
'heartbeat.interval.ms': 15000,
'get.min.bytes': 1024, # Wait for 1KB before returning
'get.max.wait.ms': 500,
'max.partition.get.bytes': 1048576, # 1MB per partition
})
consumer.subscribe(['orders.validated'], on_assign=on_partition_assign)
def process_batch():
while True:
msg = consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
raise KafkaException(msg.error())
try:
event = deserialize(msg.value())
process_event(event)
# Commit only after successful processing
consumer.commit(message=msg, asynchronous=False)
except ProcessingError as e:
publish_to_dlq(msg, e)
consumer.commit(message=msg, asynchronous=False)def publish_to_dlq(original_msg, error):
dlq_producer.produce(
topic=f"{original_msg.topic()}.dlq",
key=original_msg.key(),
value=original_msg.value(),
headers={
'original-topic': original_msg.topic(),
'original-partition': str(original_msg.partition()),
'original-offset': str(original_msg.offset()),
'error-message': str(error),
'error-timestamp': datetime.utcnow().isoformat(),
'retry-count': '0',
}
)DataStream<OrderEvent> orders = env
.fromSource(kafkaSource, WatermarkStrategy
.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((event, ts) -> event.getTimestamp()),
"orders-source");
// Tumbling window: Fixed-size, non-overlapping
DataStream<RevenueSummary> hourlyRevenue = orders
.keyBy(OrderEvent::getRegion)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.allowedLateness(Time.minutes(5))
.sideOutputLateData(lateOutputTag)
.aggregate(new RevenueAggregator(), new RevenueWindowFunction());
// Sliding window: Fixed-size, overlapping
DataStream<MovingAverage> movingAvg = orders
.keyBy(OrderEvent::getProductId)
.window(SlidingEventTimeWindows.of(Time.minutes(30), Time.minutes(5)))
.aggregate(new AverageAggregator());
// Session window: Gap-based, dynamic size
DataStream<UserSession> sessions = clickstream
.keyBy(ClickEvent::getUserId)
.window(EventTimeSessionWindows.withGap(Time.minutes(15)))
.process(new SessionWindowFunction());| Strategy | Use When | Window Size | Overlap | Late Data |
|---|---|---|---|---|
| Tumbling | Regular aggregation periods | Fixed | No | Allowed lateness |
| Sliding | Moving averages, trends | Fixed | Yes | Allowed lateness |
| Session | User activity grouping | Dynamic | No | Gap-based |
| Global | Custom trigger logic | Unbounded | N/A | Trigger-dependent |
| Count | Fixed-count batches | N items | No | N/A |
// Enable exactly-once checkpointing
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(120000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setExternalizedCheckpointRetention(
ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
// Kafka sink with exactly-once
KafkaSink<String> sink = KafkaSink.<String>builder()
.setBootstrapServers("broker:9092")
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic("output-topic")
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("flink-job-v1")
.build();from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, from_json, avg, count
from pyspark.sql.types import StructType, StringType, DoubleType, TimestampType
spark = SparkSession.builder \
.config("spark.sql.streaming.checkpointLocation", "/checkpoints/order-agg") \
.config("spark.sql.shuffle.partitions", 20) \
.getOrCreate()
schema = StructType() \
.add("order_id", StringType()) \
.add("amount", DoubleType()) \
.add("region", StringType()) \
.add("event_time", TimestampType())
orders = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "broker:9092") \
.option("subscribe", "orders") \
.option("startingOffsets", "latest") \
.option("maxOffsetsPerTrigger", 100000) \
.load() \
.select(from_json(col("value").cast("string"), schema).alias("data")) \
.select("data.*") \
.withWatermark("event_time", "10 minutes")
# Windowed aggregation
revenue_by_region = orders \
.groupBy(
window(col("event_time"), "1 hour", "15 minutes"),
col("region")
) \
.agg(
count("order_id").alias("order_count"),
avg("amount").alias("avg_amount")
)
query = revenue_by_region.writeStream \
.outputMode("update") \
.format("delta") \
.option("checkpointLocation", "/checkpoints/revenue") \
.trigger(processingTime="30 seconds") \
.start("/output/revenue_by_region")Symptoms of backpressure:
- Consumer lag growing continuously
- Processing latency increasing over time
- Memory usage climbing in stream processors
- Checkpoint durations increasing (Flink)
Diagnosis checklist:
[ ] Check consumer group lag: kafka-consumer-groups --describe --group <id>
[ ] Monitor Flink backpressure metrics: taskmanager.job.task.isBackPressured
[ ] Check Spark streaming batch duration vs trigger interval
[ ] Review GC logs for memory pressure
[ ] Profile serialization/deserialization overhead# Strategy 1: Rate limiting at source
kafka_source_config = {
'maxOffsetsPerTrigger': 50000, # Spark: limit records per trigger
'get.max.bytes': 5242880, # Kafka consumer: limit get size
}
# Strategy 2: Async processing with bounded queue
import asyncio
from asyncio import Semaphore
semaphore = Semaphore(100) # Max 100 concurrent operations
async def process_with_backpressure(event):
async with semaphore:
result = await process_event_async(event)
return result
# Strategy 3: Spillable state backend (Flink)
# flink-conf.yaml:
# state.backend: rocksdb
# state.backend.rocksdb.memory.managed: true
# state.backend.rocksdb.memory.fixed-per-slot: 256mb
# Strategy 4: Dynamic scaling
# Monitor lag, scale consumers when threshold exceeded
def autoscale_consumers(current_lag, threshold=100000):
if current_lag > threshold * 3:
scale_consumer_group(replicas=current_replicas * 2)
elif current_lag > threshold:
scale_consumer_group(replicas=current_replicas + 1)
elif current_lag < threshold * 0.1:
scale_consumer_group(replicas=max(1, current_replicas - 1))from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
schema_registry = SchemaRegistryClient({'url': '[reference URL]'})
# Schema evolution compatibility rules:
# BACKWARD - new schema can read old data (safe for consumers first)
# FORWARD - old schema can read new data (safe for producers first)
# FULL - both backward and forward compatible
# NONE - no compatibility checking
# Set compatibility level
schema_registry.set_compatibility(
subject_name="orders-value",
level="BACKWARD"
)
# Safe schema changes (backward compatible):
# - Add optional field with default
# - Remove field that had a default
# - Change field from specific type to union with null
# Unsafe schema changes:
# - Remove required field
# - Change field type
# - Rename field (no concept of rename in Avro)producer_metrics:
- record-send-rate # Records/sec being produced
- record-error-rate # Failed sends/sec
- request-latency-avg # Average broker response time
- batch-size-avg # Bytes per batch
- buffer-available-bytes # Remaining buffer capacity
consumer_metrics:
- records-lag-max # Maximum partition lag
- records-consumed-rate # Records/sec consumed
- commit-latency-avg # Offset commit time
- rebalance-rate # Consumer group rebalances
flink_metrics:
- numRecordsInPerSecond # Input throughput
- numRecordsOutPerSecond # Output throughput
- currentInputWatermark # Watermark progress
- lastCheckpointDuration # Checkpoint time
- isBackPressured # Backpressure indicator
alerts:
- consumer_lag > 100000 for 5m # Falling behind
- checkpoint_duration > 60s # State too large
- error_rate > 0.01 # > 1% errors
- rebalance_count > 3 in 10m # Unstable group# Consumer lag trend
sum(kafka_consumer_group_lag) by (group, topic)
# Processing throughput
rate(flink_taskmanager_job_task_numRecordsInPerSecond[5m])
# End-to-end latency (event time to processing time)
histogram_quantile(0.99,
rate(stream_processing_latency_seconds_bucket[5m])
)
# Backpressure ratio
avg(flink_taskmanager_job_task_isBackPressured) by (task_name)| Guarantee | How | Trade-off | Use When |
|---|---|---|---|
| At-most-once | Auto-commit before processing | Fastest, may lose data | Metrics, logs, non-critical |
| At-least-once | Commit after processing | Duplicates possible | Most use cases + idempotent sink |
| Exactly-once | Transactions or idempotent writes | Slowest, most complex | Financial, inventory, billing |
# Pattern: Deduplication with idempotency key
class IdempotentProcessor:
def __init__(self, redis_client):
self.redis = redis_client
self.dedup_ttl = 86400 # 24 hours
def process(self, event):
idempotency_key = f"processed:{event.topic}:{event.partition}:{event.offset}"
# Check if already processed
if self.redis.exists(idempotency_key):
return # Skip duplicate
# Process within transaction
with db.begin():
result = handle_event(event)
db.execute(insert_result(result))
# Mark as processed
self.redis.setex(idempotency_key, self.dedup_ttl, "1")ANTI-PATTERN FIX
------------------------------------------------------------------
Synchronous HTTP calls in stream Use async I/O or batch lookups
processing
Unbounded state accumulation Set state TTL, use windowed state
Processing one record at a time Use micro-batching or bulk APIs
Ignoring late data Configure watermarks + allowed
lateness + side outputs
Committing offsets before processing Commit after processing succeeds
No dead letter queue Always route failed records to DLQ
Hardcoded parallelism Size based on partition count and
throughput requirements
No schema management Use Schema Registry from day oneUse this skill when:
Designing or implementing streaming data engineer solutions
Reviewing or improving existing streaming data engineer approaches
Making architectural or implementation decisions about streaming data engineer
Learning streaming data engineer patterns and best practices
Troubleshooting streaming data engineer-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 Data Engineer 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 data engineer for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended streaming data engineer approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Legacy system integration: When streaming data engineer 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
496 lines · 1,761 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 Data Engineer bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for