Apache Kafka cho Kỹ sư Dữ liệu: Streaming, Partitions và Câu hỏi Phỏng vấn

Apache Kafka nằm ở trung tâm của hầu hết các stack data engineering hiện đại, xử lý hàng nghìn tỷ sự kiện mỗi ngày tại các tổ chức ở mọi quy mô. Với Kafka 4.x chạy hoàn toàn trên KRaft (không còn phụ thuộc ZooKeeper), nền tảng này trở nên đơn giản hơn để vận hành mà vẫn giữ nguyên các đảm bảo đã đưa nó trở thành chuẩn mực ngành cho các pipeline dữ liệu thời gian thực.

Kiến trúc Streaming của Kafka cho Pipeline Dữ liệu

Kafka hoạt động như một distributed commit log. Producer ghi tiếp các bản ghi vào cuối các partition của topic, và consumer đọc các bản ghi đó theo thứ tự. Thiết kế append-only này cho phép cả streaming thời gian thực lẫn replay batch từ bất kỳ offset nào, giúp Kafka đặc biệt phù hợp với các workload data engineering yêu cầu cả hai mô hình.

Một Kafka cluster bao gồm các broker (server), topic (kênh logic) và partition (shard vật lý của một topic). Mỗi partition là một chuỗi bản ghi có thứ tự và bất biến. Một broker đóng vai trò partition leader, xử lý mọi thao tác đọc và ghi, trong khi các follower replica duy trì bản sao để đảm bảo fault tolerance.

Thuộc tính kiến trúc cốt lõi: thứ tự được đảm bảo trong phạm vi một partition, không phải xuyên suốt các partition. Ràng buộc duy nhất này chi phối phần lớn các quyết định thiết kế Kafka trong pipeline dữ liệu.

# docker-compose.yml — Kafka 4.x KRaft cluster (no ZooKeeper)
services:
  kafka-1:
    image: apache/kafka:4.2.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_NUM_PARTITIONS: 6
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
    ports:
      - "9092:9092"

Cấu hình này khởi động một broker chế độ KRaft đồng thời cũng đóng vai trò controller. Tham số KAFKA_CONTROLLER_QUORUM_VOTERS thay thế những gì ZooKeeper từng quản lý: bầu chọn leader, metadata topic và quản lý thành viên cluster.

Chiến lược Partition Định hình Hiệu năng Pipeline

Partition quyết định mức độ song song, thứ tự và throughput. Việc lựa chọn số lượng partition và chiến lược key đúng đắn ảnh hưởng trực tiếp đến độ tin cậy của pipeline và khả năng mở rộng consumer.

Hướng dẫn chọn số lượng partition:

• Khởi đầu bằng số consumer dự kiến xử lý topic đồng thời
• Mỗi partition có thể duy trì throughput ghi khoảng 10 MB/giây trên phần cứng hiện đại
• Nhiều partition hơn làm tăng nhẹ độ trễ end-to-end (nhiều cuộc bầu chọn leader hơn, nhiều file handle hơn)
• Kafka 4.x xử lý hàng nghìn partition mỗi broker hiệu quả nhờ các cải tiến metadata KRaft

Lựa chọn partition key quyết định bản ghi nào rơi vào partition nào. Các bản ghi có cùng key luôn vào cùng một partition, bảo toàn thứ tự cho key đó.

# partition_strategy.py — Choosing the right partition key
from confluent_kafka import Producer
import json

def create_producer():
    return Producer({
        'bootstrap.servers': 'kafka-1:9092',
        'acks': 'all',
        'enable.idempotence': True,
        'max.in.flight.requests.per.connection': 5
    })

def publish_order_event(producer, event):
    # Key by customer_id: all events for one customer stay ordered
    key = str(event['customer_id']).encode('utf-8')
    value = json.dumps(event).encode('utf-8')
    producer.produce(
        topic='order-events',
        key=key,
        value=value,
        headers=[('source', b'order-service'), ('version', b'2')]
    )
    producer.flush()

def publish_clickstream(producer, event):
    # Key by session_id: ordering within a session matters
    # NOT by user_id — one user may have multiple sessions
    key = str(event['session_id']).encode('utf-8')
    value = json.dumps(event).encode('utf-8')
    producer.produce(
        topic='clickstream',
        key=key,
        value=value
    )

Việc chọn customer_id so với session_id làm partition key phản ánh các yêu cầu thứ tự khác nhau. Order events cần thứ tự theo từng khách hàng để duy trì tính nhất quán giao dịch. Clickstream events cần thứ tự theo từng phiên để tái hiện chính xác hành trình người dùng.

Consumer Groups và Quản lý Offset

Consumer group cho phép xử lý song song một topic. Kafka gán mỗi partition cho đúng một consumer trong group, vì vậy mức song song tối đa bằng số lượng partition.

Quản lý offset quyết định ngữ nghĩa exactly-once so với at-least-once. Kafka lưu trữ offset đã commit trong topic nội bộ __consumer_offsets. Thời điểm commit offset so với thời điểm xử lý quyết định đảm bảo phân phối.

# consumer_pipeline.py — Consumer group with manual offset management
from confluent_kafka import Consumer, KafkaError
import json

def create_consumer(group_id: str):
    return Consumer({
        'bootstrap.servers': 'kafka-1:9092',
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,  # Manual commit for at-least-once
        'max.poll.interval.ms': 300000,
        'session.timeout.ms': 45000,
        'isolation.level': 'read_committed'  # Only read committed transactions
    })

def process_batch(consumer: Consumer, batch_size: int = 500):
    """Process records in micro-batches for throughput."""
    consumer.subscribe(['order-events'])
    buffer = []

    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 Exception(msg.error())

        record = json.loads(msg.value().decode('utf-8'))
        buffer.append(record)

        if len(buffer) >= batch_size:
            # Process the full batch (write to warehouse, etc.)
            write_to_warehouse(buffer)
            # Commit AFTER successful processing
            consumer.commit(asynchronous=False)
            buffer.clear()

def write_to_warehouse(records: list):
    # Insert batch into data warehouse (BigQuery, Snowflake, etc.)
    pass

Đặt enable.auto.commit thành False và chỉ commit sau khi write_to_warehouse thành công đảm bảo phân phối at-least-once. Nếu consumer crash sau khi xử lý nhưng trước khi commit, các bản ghi sẽ được phân phối lại khi khởi động lại. Đối với pipeline đưa dữ liệu vào warehouse, đây thường là sự đánh đổi đúng đắn: ghi idempotent vào warehouse xử lý các bản trùng lặp.

Kafka Connect cho Tích hợp Pipeline Dữ liệu

Kafka Connect cung cấp một framework có khả năng mở rộng để di chuyển dữ liệu giữa Kafka và các hệ thống bên ngoài mà không cần viết mã producer/consumer tùy chỉnh. Source connector đưa dữ liệu vào Kafka; sink connector đẩy dữ liệu ra ngoài.

{
  "name": "postgres-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-primary",
    "database.port": "5432",
    "database.user": "replication_user",
    "database.dbname": "production",
    "topic.prefix": "cdc",
    "table.include.list": "public.orders,public.customers,public.products",
    "slot.name": "debezium_slot",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication",
    "snapshot.mode": "initial",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.public\\.(.*)",
    "transforms.route.replacement": "warehouse.$1"
  }
}

Debezium CDC connector này bắt mọi INSERT, UPDATE và DELETE từ các bảng PostgreSQL và xuất bản chúng dưới dạng sự kiện Kafka. Transform RegexRouter đổi tên topic từ cdc.public.orders thành warehouse.orders, giữ namespace pipeline gọn gàng. Kết hợp với BigQuery hoặc Snowflake sink connector, thiết lập này tạo ra một pipeline CDC tự động hoàn toàn mà không cần mã tùy chỉnh.

Ngữ nghĩa Exactly-Once trong Pipeline Kafka

Kafka hỗ trợ ngữ nghĩa exactly-once (EOS) thông qua idempotent producer và transactional API. Đối với pipeline data engineering, EOS ngăn các bản ghi trùng lặp trong các hệ thống downstream mà không cần logic deduplication.

Ba thành phần khiến EOS hoạt động:

1. Idempotent producer: Mỗi producer nhận một Producer ID duy nhất. Broker khử trùng lặp các lần thử lại bằng số thứ tự, do đó các lần thử lại mạng không bao giờ tạo ra bản trùng.
2. Transaction: Một producer có thể ghi nguyên tử vào nhiều partition và commit offset trong cùng một transaction. Hoặc tất cả các thao tác ghi thành công, hoặc không có thao tác nào diễn ra.
3. Cô lập read_committed: Consumer được cấu hình với isolation.level=read_committed chỉ thấy các bản ghi từ những transaction đã commit.

# exactly_once_pipeline.py — Transactional consume-transform-produce
from confluent_kafka import Consumer, Producer
import json

def transactional_pipeline():
    consumer = Consumer({
        'bootstrap.servers': 'kafka-1:9092',
        'group.id': 'etl-transformer',
        'enable.auto.commit': False,
        'isolation.level': 'read_committed'
    })

    producer = Producer({
        'bootstrap.servers': 'kafka-1:9092',
        'transactional.id': 'etl-transformer-001',
        'acks': 'all',
        'enable.idempotence': True
    })

    producer.init_transactions()
    consumer.subscribe(['raw-events'])

    while True:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue

        # Begin atomic transaction
        producer.begin_transaction()
        try:
            raw = json.loads(msg.value())
            enriched = transform(raw)

            # Write transformed record to output topic
            producer.produce(
                'enriched-events',
                key=msg.key(),
                value=json.dumps(enriched).encode('utf-8')
            )

            # Commit consumer offset within the same transaction
            producer.send_offsets_to_transaction(
                consumer.position(consumer.assignment()),
                consumer.consumer_group_metadata()
            )
            producer.commit_transaction()
        except Exception:
            producer.abort_transaction()

def transform(record: dict) -> dict:
    # Apply business logic transformations
    record['processed_at'] = '2026-04-20T00:00:00Z'
    record['pipeline_version'] = '2.1'
    return record

Tham số transactional.id cho phép broker fence các zombie producer. Nếu một consumer crash và một instance mới khởi động với cùng transactional.id, broker sẽ hủy bỏ mọi transaction còn treo từ instance cũ, ngăn ngừa output trùng lặp.

Share Groups: Ngữ nghĩa Hàng đợi của Kafka 4.2

Kafka 4.2 giới thiệu Share Groups sẵn sàng cho production, mang ngữ nghĩa hàng đợi tin nhắn truyền thống vào Kafka. Khác với consumer group, nơi mỗi partition được gán cho đúng một consumer, Share Groups cho phép nhiều consumer xử lý các bản ghi từ cùng một partition một cách độc lập.

| Đặc điểm | Consumer Groups | Share Groups | |---|---|---| | Gán partition | Độc quyền (một consumer trên mỗi partition) | Chia sẻ (nhiều consumer trên mỗi partition) | | Đảm bảo thứ tự | Bảo toàn thứ tự trong partition | Không có đảm bảo thứ tự | | Tình huống sử dụng | Luồng sự kiện có thứ tự | Phân phối tác vụ, hàng đợi công việc | | Acknowledgement | Dựa trên offset (commit position) | Theo từng bản ghi (acknowledge riêng lẻ) | | Mức song song tối đa | Bị giới hạn bởi số partition | Bị giới hạn bởi số consumer |

Share Groups giải quyết một hạn chế lâu nay: mở rộng consumer vượt quá số partition. Đối với các workload như phân phối thông báo hoặc phân phối job batch nơi thứ tự không quan trọng, Share Groups loại bỏ nhu cầu sử dụng hệ thống hàng đợi bên ngoài song song với Kafka.

Câu hỏi Phỏng vấn Kafka cho Kỹ sư Dữ liệu

Các buổi phỏng vấn kỹ thuật cho vị trí data engineer thường xuyên kiểm tra kiến thức Kafka. Các câu hỏi sau đây bao quát những khái niệm thường được đánh giá nhất.

Q: Kafka đảm bảo thứ tự tin nhắn như thế nào? Thứ tự chỉ được đảm bảo trong phạm vi một partition duy nhất. Mọi bản ghi có cùng partition key đi vào cùng một partition và được ghi tiếp theo trình tự. Giữa các partition, không tồn tại thứ tự nào. Lựa chọn đúng partition key là cơ chế chính để kiểm soát thứ tự trong pipeline Kafka.

Q: Điều gì xảy ra khi một consumer trong group bị lỗi? Group coordinator phát hiện lỗi qua heartbeat bị bỏ lỡ (kiểm soát bởi session.timeout.ms). Nó kích hoạt rebalance, gán lại các partition của consumer bị lỗi cho các thành viên còn lại trong group. Trong quá trình rebalance, việc tiêu thụ tạm dừng trong thời gian ngắn. Cooperative sticky rebalancing (mặc định trong Kafka 4.x) giảm thiểu gián đoạn bằng cách chỉ gán lại các partition bị ảnh hưởng.

Q: Giải thích sự khác biệt giữa acks=1 và acks=all. Với acks=1, broker xác nhận thao tác ghi sau khi leader replica lưu trữ bản ghi. Với acks=all, broker chờ đến khi tất cả các in-sync replica (ISR) xác nhận thao tác ghi. acks=all kết hợp với min.insync.replicas=2 đảm bảo không mất dữ liệu nếu bất kỳ broker đơn lẻ nào hỏng, đánh đổi bằng độ trễ cao hơn một chút.

Q: Bạn sẽ thiết kế một pipeline CDC sử dụng Kafka như thế nào? Bắt các thay đổi từ database nguồn sử dụng Debezium (cho PostgreSQL, MySQL) hoặc một CDC connector native. Xuất bản các sự kiện thay đổi tới các topic Kafka được phân partition theo khóa chính. Sử dụng sink connector (BigQuery Sink, S3 Sink) để load các thay đổi vào kho dữ liệu phân tích. Đặt isolation.level=read_committed trên consumer để tránh đọc các transaction database chưa commit. Sử dụng schema registry để quản lý schema Avro hoặc Protobuf cho các sự kiện thay đổi.

Q: ISR là gì và tại sao nó quan trọng? ISR (In-Sync Replicas) là tập hợp các replica đã đồng bộ đầy đủ với partition leader. Chỉ các thành viên ISR mới đủ điều kiện để được bầu làm leader. Tham số min.insync.replicas định nghĩa bao nhiêu replica phải xác nhận một thao tác ghi trước khi nó được coi là đã commit. Nếu ISR thu nhỏ xuống dưới min.insync.replicas, broker từ chối các thao tác ghi thay vì chấp nhận rủi ro mất dữ liệu.

Để chuẩn bị thêm cho phỏng vấn data engineering, bộ sưu tập câu hỏi phỏng vấn data engineering bao quát các chủ đề rộng hơn bao gồm mẫu pipeline ETL/ELT và mô hình hóa dữ liệu.

Kết luận

• Kafka 4.x với KRaft loại bỏ hoàn toàn ZooKeeper, giảm chi phí vận hành cho các đội data engineering
• Lựa chọn partition key quyết định đảm bảo thứ tự: chọn key dựa trên yêu cầu xử lý downstream, không dựa trên sự tiện lợi
• Commit offset thủ công với enable.auto.commit=False cho phép phân phối at-least-once; transactional API cung cấp ngữ nghĩa exactly-once cho các pipeline consume-transform-produce
• Kafka Connect với Debezium cung cấp CDC sẵn sàng production mà không cần mã consumer tùy chỉnh
• Share Groups (Kafka 4.2) bổ sung ngữ nghĩa hàng đợi cho các workload phân phối tác vụ nơi thứ tự không bắt buộc
• Việc mở rộng consumer group bị giới hạn bởi số partition: lập kế hoạch số lượng partition dựa trên nhu cầu song song consumer cao điểm