Apache Kafka Là Gì? Event Streaming Platform Cho Big Data

Apache Kafka là một nền tảng streaming phân tán mã nguồn mở được phát triển bởi LinkedIn và hiện là project của Apache Software Foundation. Kafka được thiết kế để xử lý luồng dữ liệu lớn theo thời gian thực với hiệu suất cao và độ tin cậy lớn.

Apache Kafka Là Gì?

Kafka là một distributed event streaming platform có khả năng:

• Publish và subscribe streams of records
• Lưu trữ streams of records một cách fault-tolerant
• Xử lý streams of records khi chúng xảy ra

Đặc điểm nổi bật:

• High throughput: Có thể xử lý hàng triệu messages mỗi giây
• Scalable: Dễ dàng mở rộng bằng cách thêm broker nodes
• Fault-tolerant: Dữ liệu được replicate nhiều lần
• Durable: Messages được persist lên disk
• Low latency: Xử lý real-time với độ trễ thấp

Kiến Trúc Apache Kafka

Các Thành Phần Chính

1. Kafka Broker

Broker là server trong Kafka cluster, chịu trách nhiệm lưu trữ và phân phối dữ liệu.

# Khởi động Kafka broker
bin/kafka-server-start.sh config/server.properties

# Kiểm tra cluster status
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe

2. Topics và Partitions

Topic là category để publish messages. Mỗi topic được chia thành nhiều partitions để scaling.

# Tạo topic với 3 partitions và replication factor 2
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic user-events \
  --partitions 3 \
  --replication-factor 2

# List tất cả topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# Xem thông tin topic
bin/kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic user-events

3. Producers

Producers gửi messages đến Kafka topics.

// Producer với Node.js và kafka-node
const kafka = require('kafka-node');
const Producer = kafka.Producer;
const client = new kafka.KafkaClient({ kafkaHost: 'localhost:9092' });
const producer = new Producer(client);

const payloads = [
  {
    topic: 'user-events',
    messages: JSON.stringify({ userId: 123, action: 'login', timestamp: Date.now() }),
    partition: 0
  }
];

producer.on('ready', () => {
  producer.send(payloads, (err, data) => {
    if (err) console.error('Error:', err);
    else console.log('Sent:', data);
  });
});

producer.on('error', (err) => {
  console.error('Producer error:', err);
});

4. Consumers và Consumer Groups

Consumers đọc messages từ topics. Consumers trong cùng group sẽ chia nhau partitions.

// Consumer với consumer group
const kafka = require('kafka-node');
const ConsumerGroup = kafka.ConsumerGroup;

const consumerOptions = {
  kafkaHost: 'localhost:9092',
  groupId: 'user-event-consumers',
  autoCommit: true,
  fetchMaxWaitMs: 1000,
  fetchMaxBytes: 1024 * 1024,
  encoding: 'utf8'
};

const consumerGroup = new ConsumerGroup(
  consumerOptions,
  ['user-events']
);

consumerGroup.on('message', (message) => {
  console.log('Received:', message.value);
  const event = JSON.parse(message.value);
  
  // Xử lý event
  processUserEvent(event);
});

consumerGroup.on('error', (err) => {
  console.error('Consumer error:', err);
});

async function processUserEvent(event) {
  try {
    switch (event.action) {
      case 'login':
        await handleUserLogin(event);
        break;
      case 'logout':
        await handleUserLogout(event);
        break;
      default:
        console.log('Unknown action:', event.action);
    }
  } catch (error) {
    console.error('Error processing event:', error);
  }
}

5. Zookeeper (Kafka 2.8+)

Zookeeper quản lý cluster metadata. Từ Kafka 2.8+, có thể chạy mà không cần Zookeeper (KRaft mode).

Kafka Streams - Xử Lý Real-time

Kafka Streams là thư viện để xây dựng ứng dụng streaming và microservices.

const { KafkaStreams } = require('kafka-streams');
const { nativeConfig: config } = require('kafka-streams').config;

config['kafka.consumer.group.id'] = 'word-count-group';
config['kafka.bootstrap.servers'] = 'localhost:9092';

const kafkaStreams = new KafkaStreams(config);

// Word count example
const stream = kafkaStreams.getKStream('text-input');

stream
  .mapJSONConvenience()
  .map((message) => message.value.toLowerCase())
  .flatMap((sentence) => sentence.split(' '))
  .filter((word) => word.length > 0)
  .countByKey(
    (word) => word,
    'count-store',
    5, // retention time in seconds
    (word, count) => {
      return { word, count };
    }
  )
  .to('word-count-output');

stream.start();

Kafka Connect - Integration

Kafka Connect là công cụ để import/export dữ liệu từ/đến Kafka.

JDBC Source Connector

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "postgres",
    "connection.password": "password",
    "table.whitelist": "users",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "db-"
  }
}

Elasticsearch Sink Connector

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "user-events",
    "connection.url": "http://localhost:9200",
    "type.name": "_doc",
    "key.ignore": "true",
    "schema.ignore": "true"
  }
}

Kafka Monitoring và Operations

1. Command Line Tools

# Kiểm tra consumer groups
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# Xem consumer group lag
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group user-event-consumers \
  --describe

# Xem offset của consumer group
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group user-event-consumers \
  --reset-offsets --to-latest --dry-run --topic user-events

2. JMX Metrics

// Enable JMX metrics trong server.properties
jmx.port=9999
jmx.enabled=true

// Monitoring với Prometheus và Grafana
// Sử dụng Kafka Exporter hoặc JMX Exporter

3. Health Checks

const kafka = require('kafka-node');
const client = new kafka.KafkaClient({ kafkaHost: 'localhost:9092' });

function checkKafkaHealth() {
  return new Promise((resolve, reject) => {
    client.loadMetadataForTopics([], (error, results) => {
      if (error) {
        reject({ status: 'unhealthy', error: error.message });
      } else {
        resolve({ 
          status: 'healthy', 
          brokers: results[1].metadata,
          timestamp: new Date().toISOString()
        });
      }
    });
  });
}

// Health check endpoint
app.get('/health/kafka', async (req, res) => {
  try {
    const health = await checkKafkaHealth();
    res.json(health);
  } catch (error) {
    res.status(503).json(error);
  }
});

Performance Tuning

Producer Configuration

const producerOptions = {
  requireAcks: 1,
  ackTimeoutMs: 100,
  partitionerType: 2, // Custom partitioner
  batch: {
    size: 16384, // 16KB batch size
    maxWaitMs: 10
  },
  compression: 'gzip', // or 'snappy', 'lz4'
  maxInFlightRequests: 5,
  metadataMaxAgeMs: 300000
};

Consumer Configuration

const consumerOptions = {
  groupId: 'high-performance-consumers',
  sessionTimeout: 15000,
  heartbeatInterval: 3000,
  maxBytesPerPartition: 1048576, // 1MB
  fetchMinBytes: 1,
  fetchMaxWaitMs: 100,
  autoCommit: false, // Manual commit for better control
  autoCommitIntervalMs: 5000
};

Security

1. SSL/TLS Encryption

const sslOptions = {
  rejectUnauthorized: false,
  ca: [fs.readFileSync('./ca-cert', 'utf8')],
  key: fs.readFileSync('./client-key', 'utf8'),
  cert: fs.readFileSync('./client-cert', 'utf8')
};

const client = new kafka.KafkaClient({
  kafkaHost: 'localhost:9093',
  sslOptions: sslOptions
});

2. SASL Authentication

const saslOptions = {
  mechanism: 'plain',
  username: 'kafka-user',
  password: 'kafka-password'
};

const client = new kafka.KafkaClient({
  kafkaHost: 'localhost:9092',
  sasl: saslOptions
});

3. ACL (Access Control Lists)

# Tạo super user
bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type users --entity-name admin \
  --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin-secret]'

# Grant permissions
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:app-user \
  --operation Read --operation Write \
  --topic user-events

So Sánh Kafka Với Các Message Broker Khác

Kafka vs RabbitMQ

Tính năng	Kafka	RabbitMQ
Throughput	Rất cao (hàng triệu msg/s)	Trung bình
Latency	Trung bình (ms)	Thấp (sub-ms)
Message retention	Có (configurable)	Xóa sau khi ack
Consumer model	Pull-based	Push-based
Routing	Đơn giản (partition key)	Phức tạp (exchange types)
Use case	Event streaming, big data	Transaction, RPC

Kafka vs Redis Streams

Tính năng	Kafka	Redis Streams
Persistence	Disk	Memory + optional disk
Scaling	Horizontal scaling	Single node
Replication	Native support	Redis Cluster
Complexity	Phức tạp	Đơn giản
Cost	Infrastructure	Memory expensive

Use Cases Phổ Biến

1. Event Sourcing

// Lưu tất cả events của user
const userEvents = [
  { type: 'USER_CREATED', data: { userId: 123, email: 'user@example.com' } },
  { type: 'USER_UPDATED', data: { userId: 123, name: 'John Doe' } },
  { type: 'USER_ACTIVATED', data: { userId: 123, timestamp: Date.now() } }
];

// Replay events để reconstruct state
function replayUserEvents(events) {
  return events.reduce((state, event) => {
    switch (event.type) {
      case 'USER_CREATED':
        return { ...state, id: event.data.userId, email: event.data.email };
      case 'USER_UPDATED':
        return { ...state, name: event.data.name };
      case 'USER_ACTIVATED':
        return { ...state, active: true, activatedAt: event.data.timestamp };
      default:
        return state;
    }
  }, {});
}

2. Log Aggregation

// Collect logs từ multiple services
const services = ['web-server', 'api-gateway', 'database', 'cache'];

services.forEach(service => {
  const consumer = kafkaStreams.getKStream(`${service}-logs`);
  
  consumer
    .mapJSONConvenience()
    .filter(log => log.level === 'ERROR' || log.level === 'WARN')
    .forEach((errorLog) => {
      // Gửi alert hoặc lưu vào database
      sendAlert(errorLog);
      saveToErrorDatabase(errorLog);
    });
  
  consumer.start();
});

3. Real-time Analytics

// Real-time user activity tracking
const userActivityStream = kafkaStreams.getKStream('user-activity');

userActivityStream
  .mapJSONConvenience()
  .map((activity) => ({
    userId: activity.userId,
    action: activity.action,
    timestamp: activity.timestamp,
    hour: new Date(activity.timestamp).getHours()
  }))
  .countByKey(
    (activity) => `${activity.hour}-${activity.action}`,
    'hourly-actions',
    3600, // 1 hour retention
    (key, count) => {
      const [hour, action] = key.split('-');
      return { hour: parseInt(hour), action, count };
    }
  )
  .forEach((result) => {
    console.log(`Hour ${result.hour}: ${result.action} = ${result.count} times`);
  });

Best Practices

1. Topic Design

// Đặt tên topic theo convention
const topicName = `${service}.${domain}.${eventType}`;
// Ví dụ: payment.service.transaction.completed

// Số partitions phù hợp với throughput
// Rule of thumb: partitions = max(expected_throughput / 10MB/s, number_of_consumers)

2. Message Format

// Sử dụng schema registry với Avro/Protobuf
const message = {
  schema: {
    type: 'record',
    name: 'UserEvent',
    fields: [
      { name: 'userId', type: 'long' },
      { name: 'eventType', type: 'string' },
      { name: 'timestamp', type: 'long' },
      { name: 'data', type: 'string' }
    ]
  },
  payload: {
    userId: 123,
    eventType: 'USER_LOGIN',
    timestamp: Date.now(),
    data: JSON.stringify({ ip: '192.168.1.1' })
  }
};

3. Idempotent Producers

const producerOptions = {
  requireAcks: -1, // Wait for all in-sync replicas
  ackTimeoutMs: 30000,
  partitionerType: 3, // Custom partitioner với idempotence
  maxInFlightRequests: 1, // Ensure ordering
  idempotent: true, // Enable idempotence
  transactionalId: 'payment-service-tx'
};

Kết Luận

Apache Kafka là một nền tảng streaming mạnh mẽ, phù hợp cho các ứng dụng cần xử lý lượng lớn dữ liệu theo thời gian thực. Với khả năng mở rộng cao, độ tin cậy lớn và ecosystem phong phú, Kafka là lựa chọn tuyệt vời cho:

• Event streaming: Xây dựng event-driven architecture
• Real-time analytics: Xử lý và phân tích dữ liệu real-time
• Log aggregation: Thu thập và xử lý logs từ nhiều nguồn
• Event sourcing: Lưu trữ và replay events
• Microservices communication: Giao tiếp bất đồng bộ giữa services

Việc hiểu rõ kiến trúc, cách cấu hình và best practices của Kafka sẽ giúp bạn xây dựng hệ thống streaming hiệu quả và đáng tin cậy.