Apache Kafka est une plateforme de streaming distribuée qui excelle dans le traitement de flux de données à haute performance. Cet article plonge dans les aspects techniques avancés de Kafka, offrant des insights pratiques pour les ingénieurs et architectes de données.
1. Configuration et Optimisation du Cluster
1.1 Installation et Configuration de Base
# Téléchargement et extraction de Kafka
wget https://downloads.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz
tar -xzf kafka_2.13-3.3.1.tgz
cd kafka_2.13-3.3.1
# Démarrage de ZooKeeper (bientôt remplacé par KRaft)
bin/zookeeper-server-start.sh config/zookeeper.properties
# Démarrage du broker Kafka
bin/kafka-server-start.sh config/server.properties1.2 Optimisation des Paramètres du Broker
# Édition du fichier server.properties
vim config/server.properties
# Paramètres clés à ajuster
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.retention.hours=168
num.partitions=8
default.replication.factor=32. Gestion Avancée des Topics et Partitions
2.1 Création et Configuration de Topics
# Création d'un topic avec des paramètres avancés
bin/kafka-topics.sh --create --topic high-throughput-topic \
--bootstrap-server localhost:9092 \
--partitions 32 \
--replication-factor 3 \
--config max.message.bytes=1000000 \
--config retention.ms=86400000
# Liste des topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# Description détaillée d'un topic
bin/kafka-topics.sh --describe --topic high-throughput-topic --bootstrap-server localhost:90922.2 Rééquilibrage des Partitions
# Génération d'un plan de réassignation
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
--topics-to-move-json-file topics-to-move.json \
--broker-list "0,1,2,3" \
--generate
# Exécution du plan de réassignation
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
--reassignment-json-file reassignment.json \
--execute
# Vérification de la progression
bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
--reassignment-json-file reassignment.json \
--verify3. Développement Avancé avec l’API Kafka
3.1 Producteur Haute Performance
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 3);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("compression.type", "snappy");
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 1000000; i++) {
producer.send(new ProducerRecord<String, String>("high-throughput-topic",
Integer.toString(i), Integer.toString(i)),
(metadata, exception) -> {
if (exception != null) {
exception.printStackTrace();
}
}
);
}
producer.close();3.2 Consommateur avec Traitement Parallèle
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "high-throughput-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("max.poll.records", 500);
props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("high-throughput-topic"));
ExecutorService executor = Executors.newFixedThreadPool(8);
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
if (!records.isEmpty()) {
executor.submit(() -> processRecords(records));
}
consumer.commitAsync();
}
private void processRecords(ConsumerRecords<String, String> records) {
for (ConsumerRecord<String, String> record : records) {
// Traitement parallèle des enregistrements
}
}4. Monitoring et Maintenance du Cluster
4.1 Collecte de Métriques JMX
# Activation de JMX
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9999
-Dcom.sun.management.jmxremote.authenticate=false
-Dcom.sun.management.jmxremote.ssl=false"
# Utilisation de jconsole pour la visualisation
jconsole localhost:99994.2 Gestion des Logs et Nettoyage
# Vérification de la taille des logs
du -sh /path/to/kafka-logs/*
# Nettoyage manuel des segments de log
bin/kafka-delete-records.sh --bootstrap-server localhost:9092 \
--offset-json-file delete-records.json
# Contenu de delete-records.json
{
"partitions": [
{
"topic": "high-throughput-topic",
"partition": 0,
"offset": 1000000
}
],
"version": 1
}5. Sécurité et Authentification
5.1 Configuration SSL/TLS
# Génération de certificats
keytool -keystore kafka.server.keystore.jks -alias localhost -validity 365 -genkey -keyalg RSA
# Configuration dans server.properties
listeners=SSL://localhost:9093
security.inter.broker.protocol=SSL
ssl.keystore.location=/path/to/kafka.server.keystore.jks
ssl.keystore.password=keystore_password
ssl.key.password=key_password5.2 Mise en Place des ACLs
# Ajout d'une ACL
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
--add --allow-principal User:Bob --operation Read --topic test-topic
# Liste des ACLs
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
--list --topic test-topicCes exemples avancés illustrent la profondeur et la complexité de la configuration et de l’utilisation d’Apache Kafka dans des environnements de production exigeants. La maîtrise de ces concepts et techniques permet de construire des systèmes de traitement de données robustes et hautement performants.
