Kafka & Kinesis Reference

Streams, shards, topics, partitions, and the producer/consumer code for both platforms.

1. Streaming Concepts

Stream processing handles unbounded, continuously arriving data — clickstreams, IoT telemetry, transactions — instead of a fixed batch. A producer writes records; a broker/service durably stores and orders them; a consumer reads and processes them, tracking its own position so it can resume after a restart.

2. Amazon Kinesis

A stream is divided into shards, each providing 1MB/sec (or 1,000 records/sec) write and 2MB/sec read. A record's partition key is hashed to pick its shard — records with the same key always land on the same shard, preserving their relative order.

3. Kinesis with boto3

import boto3, json

kinesis = boto3.client("kinesis", region_name="us-east-1")

# Producer: put a single record
kinesis.put_record(
    StreamName="orders-stream",
    Data=json.dumps({"order_id": 123, "amount": 49.99}).encode("utf-8"),
    PartitionKey="order_123"
)

# Producer: batch put
kinesis.put_records(
    StreamName="orders-stream",
    Records=[
        {"Data": json.dumps({"order_id": i}).encode(), "PartitionKey": str(i)}
        for i in range(10)
    ]
)

# Consumer: read from a shard
shard_id = kinesis.describe_stream(StreamName="orders-stream")[
    "StreamDescription"]["Shards"][0]["ShardId"]

shard_iterator = kinesis.get_shard_iterator(
    StreamName="orders-stream", ShardId=shard_id,
    ShardIteratorType="TRIM_HORIZON"
)["ShardIterator"]

response = kinesis.get_records(ShardIterator=shard_iterator, Limit=100)
for record in response["Records"]:
    print(json.loads(record["Data"]))

4. Apache Kafka

Kafka is a distributed log: topics are split into partitions spread across brokers. Each partition is an ordered, append-only log; order is guaranteed only within a partition. Consumer groups let multiple consumers split a topic's partitions between them for parallel processing, with Kafka tracking each group's committed offset per partition.

• Replication factor — copies each partition across brokers for durability, mirroring HDFS's replication concept.
• Retention — messages persist for a configured time/size regardless of whether they've been consumed — unlike a queue, Kafka doesn't delete on read.
• Keys — a message key determines its partition, the same role Kinesis's partition key plays.

5. Kafka with Python

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)
producer.send("orders", key=b"order_123", value={"order_id": 123, "amount": 49.99})
producer.flush()

# Consumer, as part of a group
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers="localhost:9092",
    group_id="orders-processor",
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    auto_offset_reset="earliest"
)
for message in consumer:
    print(message.partition, message.offset, message.value)

6. Kinesis vs Kafka

7. Offsets & Delivery Guarantees

An offset is a record's position within its partition/shard. Consumers track the offset they've processed up to, so restarting resumes correctly instead of reprocessing or skipping.

• Auto-commit — the client library commits offsets on a timer, regardless of whether processing actually finished; simplest, but can lose or duplicate records on a crash.
• Manual commit — commit only after a record (or batch) is fully processed — the standard way to get at-least-once delivery.
• Exactly-once — requires pairing at-least-once delivery with an idempotent write at the sink (a keyed upsert), the same principle covered in the Incremental Load reference doc.

consumer = KafkaConsumer(
    "orders", bootstrap_servers="localhost:9092",
    group_id="orders-processor", enable_auto_commit=False
)
for message in consumer:
    process(message.value)
    consumer.commit()  # only after processing succeeds

8. Scaling: Resharding & Rebalancing

• Kinesis resharding — split a hot shard into two, or merge two under-used shards into one, to match write throughput. Resharding is an explicit, occasional operation, not automatic (On-Demand mode capacity scales automatically instead).
• Kafka partition rebalancing — adding partitions to a topic redistributes future keys' assignment; existing messages don't move, so partition count is usually set generously upfront rather than grown casually.
• Consumer group rebalancing — when a consumer joins or leaves a group, Kafka reassigns partitions among the remaining members; this pauses processing briefly during the rebalance.

9. Firehose Transforms & ksqlDB

Kinesis Data Firehose can invoke a Lambda function to transform each record before delivery (e.g. flattening JSON, filtering bad records) without standing up a separate consumer:

import base64, json

def handler(event, context):
    output = []
    for record in event["records"]:
        payload = json.loads(base64.b64decode(record["data"]))
        payload["processed_at"] = context.aws_request_id
        output.append({
            "recordId": record["recordId"],
            "result": "Ok",
            "data": base64.b64encode(json.dumps(payload).encode()).decode()
        })
    return {"records": output}

Kafka's equivalent for lightweight stream SQL is ksqlDB — declares continuous queries over topics without writing a full consumer application:

CREATE STREAM high_value_orders AS
  SELECT * FROM orders
  WHERE amount > 1000
  EMIT CHANGES;

10. Monitoring & Lag

Consumer lag — the gap between the latest offset written and the offset a consumer has processed — is the single most important streaming health metric: growing lag means consumers can't keep up with producers.

• Kafka — kafka-consumer-groups.sh --describe or a metrics exporter reports per-partition lag.
• Kinesis — CloudWatch's GetRecords.IteratorAgeMilliseconds is the equivalent signal; a rising value means a consumer is falling behind.
• Throttling — watch WriteProvisionedThroughputExceeded (Kinesis) or produce-request error rates (Kafka) as a sign shard/partition count needs to grow.

11. Apache NiFi

Apache NiFi moves data between systems. You build a flow visually by wiring together processors, each doing one step. It is built for getting data from A to B and reshaping it on the way, not for heavy analytics.

A minimal flow: CSV file to Kafka

CSV dropped in input/
  → GetFile              (picks up the file)
  → PublishKafkaRecord   (parses CSV, converts each row to JSON, sends to topic "orders")

Do you need NiFi if Spark already reads Kafka?

No. If data already arrives in Kafka, Spark reads it directly and NiFi is one more system to run. NiFi earns its place in front of Kafka, for sources that cannot publish to Kafka themselves.

Alternatives to NiFi: Kafka Connect (ready-made connectors for S3, JDBC, MongoDB and more, configured in JSON; the usual choice in Kafka-centric stacks) or small producer services for one or two simple sources.

NiFi + Spark integration patterns

Rule of thumb: NiFi connects systems and moves data; Spark does the heavy computation. Skip NiFi when every source can already write to Kafka.

12. Kafka or NiFi as a Spark Source?

Kafka, clearly. Spark should always read from Kafka; NiFi is not an alternative streaming source, at most it feeds Kafka.

[sources] → (NiFi / Kafka Connect, optional) → Kafka → Spark Structured Streaming → sink
              ingestion                       buffer     processing

13. Kafka CLI & Web UIs

Kafka ships shell tools in its bin/ directory. In Docker, run them with docker exec against the internal listener (e.g. kafka:29092); from the host use the external one (e.g. localhost:9092). In Git Bash on Windows, prefix with MSYS_NO_PATHCONV=1 to stop path rewriting.

# topics
kafka-topics.sh --bootstrap-server kafka:29092 --list
kafka-topics.sh --bootstrap-server kafka:29092 --describe --topic txn.upi
kafka-get-offsets.sh --bootstrap-server kafka:29092 --topic txn.upi

# consume from the beginning, showing keys and offsets
kafka-console-consumer.sh --bootstrap-server kafka:29092 --topic txn.upi \
  --from-beginning --property print.key=true --property print.offset=true

# produce (add these properties to send KEY|{json})
kafka-console-producer.sh --bootstrap-server kafka:29092 --topic txn.upi \
  --property parse.key=true --property key.separator="|"

Kafka web UIs

Kafbat UI (Apache 2.0) began as Provectus's "UI for Apache Kafka" and was forked by its core maintainers in 2023. It is a Spring Boot (WebFlux) backend with a React/TypeScript frontend, and is itself just a Kafka client:

• Topic and broker screens use the AdminClient.
• Message browsing uses a KafkaConsumer with assign() / seek() rather than a consumer group.
• The Produce button uses a KafkaProducer.
• Consumer lag = latest offset − committed group offset, per partition.

Alternatives: Redpanda Console, AKHQ, Confluent Control Center, Conduktor.

14. Lambda Architecture: Flink + Databricks

Lambda runs two pipelines over the same raw data: a speed layer for immediate results and a batch layer that recomputes complete, corrected history. A common banking setup uses Flink for speed and PySpark on Databricks for batch:

REAL-TIME EVENTS                         FILE DROPS (S3 landing, business_date=…)
cards · UPI · ATM · app                  core-banking EOD · settlement · masters
        │                                                   │
        ▼                                                   │
      Kafka ──► FLINK speed layer (ms) ──► serving DB / alerts topic
        │            ▲ reference data (compacted topic / Redis)
        └─► S3 raw sink                                     │
                │                                           ▼
                └──────► DATABRICKS batch layer (nightly job)
                         BRONZE → SILVER → GOLD
                         reconciliation · Customer 360 · regulatory reports
                         → publishes reference data back to Flink

Design points

• Feed both layers the same raw events, with a schema registry and shared business keys.
• Duplicated logic is Lambda's weakness: use config-driven rules, SQL where possible, shared rule libraries, and one owning layer per metric.
• Reconcile daily: counts, amounts and alerts per day and channel, speed vs batch, within tolerances.
• Exactly-once: Flink checkpoints + transactional sinks; Databricks checkpoints + MERGE or partition overwrite.
• Late data: Flink side outputs; batch reprocesses recent business dates.

Where a nightly S3 file job fits

Files arrive with a contract and a control file (row count, totals, checksum). The job: check arrivals (SLA alert) → validate control totals and schema, quarantine bad files → Auto Loader availableNow into bronze → silver MERGE / SCD2 → join with the day's events → reconcile real-time vs settled → data-quality gate → gold → publish reference data to Flink → archive and notify. Keep it idempotent per business_date.

15. Kappa vs Lambda

LAMBDA:
Sources ─┬─► BATCH layer (complete) ──► batch views ─────┐
         └─► SPEED layer (fast)     ──► real-time views ─┴─► SERVING (merge)

KAPPA:
Sources → Kafka (replayable log) ──► ONE stream processor ──► serving tables
reprocess = run v2 from offset 0 into new tables, then switch readers (blue/green)

Kappa needs a replayable source (long Kafka retention, tiered storage, or a Delta bronze table), event-time processing, idempotent or versioned sinks, and deterministic logic.

Lakehouses blur the line: a Delta or Iceberg table can be read as a batch or a stream, so the same code does incremental and full-refresh processing.

16. Watermarks in Flink & Spark

A watermark says how far event time has progressed: "all events up to T have probably arrived". With 5 seconds of allowed out-of-orderness, watermark = max event time seen − 5 s. It decides when windows emit, which late events are dropped, and when state is cleaned up.

-- Flink SQL
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND

SELECT window_start, product, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, product;

# PyFlink DataStream
WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)) \
    .with_timestamp_assigner(OrderTime()) \
    .with_idleness(Duration.of_seconds(30))

# Spark Structured Streaming
df.withWatermark("event_time", "5 seconds")

• Idle partitions: Flink's watermark is the minimum across source partitions, so one idle partition stalls it. Set with_idleness or table.exec.source.idle-timeout.
• Spark advances one global watermark per micro-batch.
• Operators without event-time windows (e.g. plain running totals) use no watermark, and their state is never cleaned up unless you set a TTL.

The cost of dropped events

Late events are dropped only from event-time operators; they still sit in Kafka and in row-by-row paths. The failure is silent: windowed totals undercount, a windowed fraud rule may see 4 transactions instead of 5, and late data is often biased toward specific sources.

• Measure: numLateRecordsDropped (Flink), numRowsDroppedByWatermark (Spark).
• Set the delay from observed lateness (e.g. p99.9 plus a margin).
• Use allowed lateness with an upsert sink, or route late events to a side output.
• Reconcile with a batch job over complete data for exact numbers.

For fraud detection, never put watermark-dropping operators on the raw transaction path: every transaction must be scored and stored. Use watermarked windows only for metrics.

17. The Kafka Data Engineer Role

Key skills

• Producers: acks=all, idempotence, batching and compression, delivery callbacks.
• Consumers: at-least-once vs exactly-once, idempotent writes, DLQs, lag, replays.
• Integration: Schema Registry, Kafka Connect, Debezium, a stream processor.
• Operations: monitoring (e.g. Prometheus/Grafana), capacity planning, managed Kafka (Confluent Cloud, MSK, Event Hubs).
• Engineering: pipelines in Git with CI/CD, tests, Terraform, runbooks.

A typical integration task

• 1. Understand the source and its volume.
• 2. Design the topic: name, partitions, key, retention.
• 3. Define and register the schema.
• 4. Build ingestion: a Debezium connector or an application producer.
• 5. Build processing, with a DLQ for bad records.
• 6. Build sinks, guarantee correctness, reconcile counts.
• 7. Monitor, secure, document and hand over.