Kafka'ya Giriş

giriiş

Apache Kafka, Java dilinde yazılmış ve gerçek zamanlı veri akışlarını işlemek için tasarlanmış, açık kaynaklı, dağıtılmış bir olay ve akış işleme platformudur. Yüksek verimlilik ve kullanılabilirlik ile doğası gereği ölçeklenebilirdir. Apache Yazılım Vakfı tarafından geliştirilen Kafka, güvenilirliği, kullanım kolaylığı ve hata toleransı nedeniyle yaygın olarak benimsenmiştir. Dünyanın en büyük kuruluşları tarafından büyük miktarda veriyi dağıtık ve verimli bir şekilde yönetmek için kullanılmaktadır.

Bu eğitimde, Apache Kafka'yı indirip kuracaksınız. Verilen betikleri kullanarak konu oluşturmayı ve silmeyi, ayrıca etkinlik göndermeyi ve almayı öğreneceksiniz. Ayrıca, aynı amaca sahip benzer projeler hakkında bilgi edinecek ve Kafka'nın diğer projelerle nasıl karşılaştırıldığını öğreneceksiniz.

Ön koşullar

• En az 4 GB RAM ve 2 CPU'ya sahip bir makine. Ubuntu Server örneğinde
• Droplet veya yerel makinenize Java 8 veya üzeri yüklü olmalıdır.

Adım 1 – Apache Kafka'yı İndirin ve Yapılandırın

Bu bölümde, Apache Kafka'yı bilgisayarınıza indirip çıkaracaksınız. Daha fazla güvenlik için, kendi kullanıcı hesabınız altında kuracaksınız. Ardından, KRaft kullanarak yapılandırıp çalıştıracaksınız.

Öncelikle, Kafka'nın altında çalışacağı ayrı bir kullanıcı oluşturun. Aşağıdaki komutu çalıştırarak kafka adında bir kullanıcı oluşturun:

sudo adduser kafka

Hesap şifreniz istenecektir. Güçlü bir şifre girin ve her alan için ENTER tuşuna basarak ek bilgi girme gereğini ortadan kaldırın.

Son olarak, belirli Kafka kullanıcısına geçin:

su kafka

Ardından, resmi indirme sayfasından Kafka sürüm paketini indireceksiniz. Yazım sırasında en son sürüm, Scala 2.13 için derlenmiş 3.7.0 idi. macOS veya Linux kullanıyorsanız, Kafka'yı curl ile indirebilirsiniz.

Kafka'yı indirmek ve /tmp dizinine yerleştirmek için şu komutu kullanın:

curl -o /tmp/kafka.tgz https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz

Sürümü ana dizindeki ~/kafka dizinine kaydedeceksiniz. Şunu çalıştırarak oluşturun:

mkdir ~/kafka

Daha sonra şunu çalıştırarak ~/kafka'ya çıkarın:

tar -xzf /tmp/kafka.tgz -C ~/kafka --strip-components=1

İndirdiğiniz arşiv, Kafka sürümüyle aynı ada sahip bir kök klasör içerdiğinden, –strip-components=1 bunu atlayacak ve içindeki her şeyi çıkaracaktır.

Bu yazının yazıldığı tarihte, Kafka 3, meta veri yönetimi için iki sistemi destekleyen son büyük sürümdü: Apache ZooKeeper ve Kafka KRaft (Kafka Raft'ın kısaltması). ZooKeeper, uygulamalar için dağıtılmış verileri koordine etmenin standart bir yolunu sağlayan ve Apache Yazılım Vakfı tarafından geliştirilen açık kaynaklı bir projedir.

Ancak Kafka 3.3 sürümünden itibaren KRaft desteği de sunuldu. KRaft, yalnızca Kafka örneklerini koordine etmek için özel olarak tasarlanmış bir sistemdir, kurulum sürecini basitleştirir ve çok daha fazla ölçeklenebilirlik sağlar. KRaft ile Kafka, yönetimsel meta verileri harici olarak tutmak yerine verilerin tüm sorumluluğunu üstlenir.

ZooKeeper desteğinin Kafka 4 ve sonrasında kaldırılması bekleniyor. Bu eğitimde, KRaft kullanarak Kafka'yı kuracaksınız.

Yeni Kafka kümeniz için benzersiz bir tanımlayıcı oluşturmanız gerekiyor. Şu anda yalnızca bir düğümden oluşuyor. Kafka'nın şu anda bulunduğu dizine gidin:

cd ~/kafka

Kafka ve KRaft yapılandırmasını config/kraft/server.properties dizininde saklarken, ZooKeeper'ın yapılandırma dosyası config/server.properties dizinindedir.

İlk kez çalıştırmadan önce, varsayılan ayarlardan bazılarını geçersiz kılmanız gerekir. Dosyayı düzenlemek için şu komutu çalıştırarak açın:

nano config/kraft/server.properties

Aşağıdaki satırları bulun:

...
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
...

log.dirs ayarı, Kafka'nın günlük dosyalarını nerede sakladığını belirtir. Varsayılan olarak, geçici de olsa yazılabilir oldukları garanti edildiğinden, bunları /tmp/kafka-logs dizininde depolar. Değeri belirtilen yolla değiştirin:

...
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/home/kafka/kafka-logs
...

Kafka için ayrı bir kullanıcı oluşturduğunuz için, günlük dizini yolunu kullanıcının ana dizininin altına yerleştireceksiniz. Eğer mevcut değilse, Kafka bunu oluşturacaktır. İşiniz bittiğinde dosyayı kaydedip kapatın.

Artık Kafka'yı yapılandırdığınıza göre, rastgele bir küme kimliği oluşturmak için aşağıdaki komutu çalıştırın:

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

Daha sonra aşağıdaki komutu çalıştırıp ID'yi girerek günlük dosyaları için depolama alanı oluşturun:

bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

Çıktı şu şekilde olacaktır:

Output
Formatting /home/kafka/kafka-logs with metadata.version 3.7-IV4.

Son olarak Kafka sunucusunu ilk kez başlatabilirsiniz:

bin/kafka-server-start.sh config/kraft/server.properties

Son çıktı şuna benzer olacaktır:

Output
...
[2024-02-26 10:38:26,889] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.DataPlaneAcceptor)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Waiting for all of the authorizer futures to be completed (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Finished waiting for all of the authorizer futures to be completed (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Waiting for all of the SocketServer Acceptors to be started (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Finished waiting for all of the SocketServer Acceptors to be started (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Transition from STARTING to STARTED (kafka.server.BrokerServer)
[2024-02-26 10:38:26,891] INFO Kafka version: 3.7.0 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,891] INFO Kafka commitId: 5e3c2b738d253ff5 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,891] INFO Kafka startTimeMs: 1708943906890 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,892] INFO [KafkaRaftServer nodeId=1] Kafka Server started (kafka.server.KafkaRaftServer)

Çıktı, Kafka'nın KRaft kullanarak başarıyla başlatıldığını ve 0.0.0.0:9092 üzerindeki bağlantıları kabul ettiğini gösteriyor.

CTRL + C'ye bastığınızda işlem sonlanır. Kafka'yı oturum açıkken çalıştırmak tercih edilmediğinden, bir sonraki adımda Kafka'yı arka planda çalıştıracak bir servis oluşturacaksınız.

Adım 2 – Kafka için bir systemd hizmeti oluşturun

Bu bölümde, Kafka'yı her zaman arka planda çalıştıracak bir systemd hizmeti oluşturacaksınız. Systemd hizmetleri sürekli olarak başlatılabilir, durdurulabilir ve yeniden başlatılabilir.

Hizmet yapılandırmasını, systemd'nin hizmetlerini depoladığı /lib/systemd/system dizinindeki code-server.service adlı bir dosyada saklayın. Dosyayı metin düzenleyicinizi kullanarak oluşturun:

sudo nano /etc/systemd/system/kafka.service

Aşağıdaki satırları ekleyin:

[Unit]
Description=kafka-server
[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/kraft/server.properties > /home/kafka/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

Burada önce hizmet açıklamasını belirtin. Ardından, [Hizmet] alanında hizmet türünü (basit, komutun basit bir şekilde çalıştırılması gerektiği anlamına gelir) tanımlayın ve çalıştırılacak komutu belirtin. Ayrıca, çalıştırılacak kullanıcının Kafka olduğunu ve Kafka'dan çıkıldığında hizmetin otomatik olarak yeniden başlatılacağını belirtin.

[Yükle] bölümü, sunucunuza giriş yapabildiğinizde sistemin bu hizmeti başlatmasını sağlar. İşlem tamamlandığında dosyayı kaydedip kapatın.

Aşağıdaki komutu çalıştırarak Kafka servisini başlatın:

sudo systemctl start kafka

Doğru şekilde başlatıldığını durumunu görüntüleyerek kontrol edin:

sudo systemctl status kafka

Aşağıdakine benzer bir çıktı göreceksiniz:

Output
● kafka.service - kafka-server
Loaded: loaded (/etc/systemd/system/kafka.service; disabled; preset: enabled)
Active: active (running) since Mon 2024-02-26 11:17:30 UTC; 2min 40s ago
Main PID: 1061 (sh)
Tasks: 94 (limit: 4646)
Memory: 409.2M
CPU: 10.491s
CGroup: /system.slice/kafka.service
├─1061 /bin/sh -c "/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/kraft/server.properties > /home/kafka/kafka/kafka.log 2>&1"
└─1062 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true "-Xlog:gc*:file=/home/kafka/kafka/bin/../logs/kaf>
Feb 26 11:17:30 kafka-test1 systemd[1]: Started kafka.service - kafka-server.

Sunucu yeniden başlatıldıktan sonra Kafka'nın otomatik olarak başlaması için aşağıdaki komutu çalıştırarak hizmetini etkinleştirin:

sudo systemctl enable kafka

Bu noktada, Kafka için bir systemd hizmeti oluşturup etkinleştirdiniz, böylece her sunucu açılışında başlatılacak. Şimdi, Kafka'da konu oluşturmayı ve silmeyi, ayrıca mevcut betikleri kullanarak metin mesajlarını nasıl oluşturup kullanacağınızı öğreneceksiniz.

Adım 3 – Güncel mesajların üretilmesi ve tüketilmesi

Artık bir Kafka sunucusu kurduğunuza göre, konular ve bunları sağlanan betikleri kullanarak nasıl yöneteceğiniz hakkında bilgi edineceksiniz. Ayrıca, bir konudan nasıl mesaj gönderip alacağınızı da öğreneceksiniz.

Olay Akışı makalesinde açıklandığı gibi, mesaj yayınlama ve alma, konularla ilişkilidir. Bir konu, mesajın ait olduğu kategoriyle ilişkilendirilebilir.

Sağlanan kafka-topics.sh betiği, Kafka'da konuları CLI aracılığıyla yönetmek için kullanılabilir. first-topic adlı bir konu oluşturmak için aşağıdaki komutu çalıştırın:

bin/kafka-topics.sh --create --topic first-topic --bootstrap-server localhost:9092

Sağlanan tüm Kafka betikleri sunucu adresini --bootstrap-server ile belirtmenizi gerektirir.

Çıktı şu şekilde olacaktır:

Output
Created topic first-topic.

Mevcut tüm konuları listelemek için –create ifadesini –list ile değiştirin:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Oluşturduğunuz konuyu görüyorsunuz:

Output
first-topic

Konuyla ilgili detaylı bilgi ve istatistiklere ulaşmak için --describe komutunu verebilirsiniz:

bin/kafka-topics.sh --describe --topic first-topic --bootstrap-server localhost:9092

Çıktı şu şekilde görünecektir:

Output
Topic: first-topic TopicId: VtjiMIUtRUulwzxJL5qVjg PartitionCount: 1 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: first-topic Partition: 0 Leader: 1 Replicas: 1 Isr: 1

İlk satır, konu adını, kimliğini ve tekrarlama faktörünü belirtir; konu yalnızca geçerli bilgisayarda mevcut olduğundan 1'dir. İkinci satır özellikle girintilidir ve konunun ilk (ve tek) bölümü hakkında bilgi gösterir. Kafka, konuyu bölümlere ayırmanıza olanak tanır; bu da bir konunun farklı bölümlerinin farklı sunuculara dağıtılabileceği ve böylece ölçeklenebilirliğin artırılabileceği anlamına gelir. Burada yalnızca bir bölüm vardır.

Artık bir konu oluşturduğunuza göre, kafka-console-producer.sh betiğini kullanarak bu konu için mesajlar üreteceksiniz. Üreticiyi başlatmak için aşağıdaki komutu çalıştırın:

bin/kafka-console-producer.sh --topic first-topic --bootstrap-server localhost:9092

Boş bir bildirim göreceksiniz:

>

Sağlayıcı SMS'inizi bekliyor. Testi girin ve ENTER'a basın. Bildirim şu şekilde görünecektir:

>test
>

Üretici artık bir sonraki mesajı bekliyor; bu, önceki mesajın Kafka'ya başarıyla iletildiği anlamına geliyor. Test için istediğiniz sayıda mesaj girebilirsiniz. Üreticiden çıkmak için CTRL+C tuşlarına basın.

Konu mesajlarını almak için bir tüketiciye ihtiyacınız var. Kafka, kafka-console-consumer.sh biçiminde basit bir tüketici sağlar. Çalıştırmak için şunu çalıştırın:

bin/kafka-console-consumer.sh --topic first-topic --bootstrap-server localhost:9092

Ancak herhangi bir çıktı alınmayacaktır. Bunun nedeni, tüketicinin konudan veri akışı yapması ve şu anda hiçbir şey üretilmemesi veya gönderilmemesidir. Tüketici başlamadan önce ürettiğiniz mesajları tüketmek için, konuyu baştan itibaren şu komutu çalıştırarak okumanız gerekir:

bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server localhost:9092

Tüketici tüm konu olaylarını tekrar oynatır ve mesajları getirir:

Outputtest
...

Oluşturucuda olduğu gibi çıkmak için CTRL+C'ye basın.

Tüketicinin gerçekten veri akışı sağladığını doğrulamak için, ayrı bir terminal oturumunda açın. İkincil bir SSH oturumu açın ve tüketiciyi varsayılan yapılandırmada çalıştırın:

bin/kafka-console-consumer.sh --topic first-topic --bootstrap-server localhost:9092

İlk oturumda, oluşturucuyu çalıştırın:

bin/kafka-console-producer.sh --topic first-topic --bootstrap-server localhost:9092

Ardından istediğiniz mesajları girin:

>second test
>third test
>

Tüketicinin eline ulaştığını hemen göreceksiniz:

Output
second test
third test

Test tamamlandıktan sonra hem üreticiyi hem de tüketiciyi sonlandırın.

İlk konuyu silmek için kafka-topics.sh'ye --delete komutunu verin:

bin/kafka-topics.sh --delete --topic first-topic --bootstrap-server localhost:9092

Çıktı alınmayacak. Konuların gerçekten silindiğini doğrulamak için listeleyebilirsiniz:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Çıktı şu şekilde olacaktır:

Output
__consumer_offsets

__Consumer_Equivalent, bir tüketicinin bir konuyu ne kadar süreyle okuduğunu depolayan Kafka'nın dahili bir konusudur.

Bu noktada, bir Kafka konusu oluşturup içinde mesajlar oluşturdunuz. Ardından, verilen betiği kullanarak mesajları tükettiniz ve sonunda gerçek zamanlı olarak aldınız. Şimdi, Kafka'nın diğer olay aracıları ve benzer yazılımlarla nasıl karşılaştırıldığını öğreneceksiniz.

Benzer mimarilerle karşılaştırma

Apache Kafka, olay akışı kullanım durumları için fiili çözüm olarak kabul edilir. Ancak Apache Pulsar ve RabbitMQ da yaygın olarak kullanılmakta ve yaklaşımlarında farklılıklar olsa da çok yönlü seçenekler olarak öne çıkmaktadır.

Mesaj kuyruğu ile olay akışı arasındaki temel fark, birincisinin temel görevinin, mesajları tüketicilere, sıralarına bakılmaksızın mümkün olan en hızlı şekilde iletmek olmasıdır. Bu tür sistemler, genellikle mesajları tüketiciler tarafından onaylanana kadar bellekte saklar. Mesaj filtreleme ve yönlendirme, tüketicilerin belirli veri kategorilerine ilgi gösterebilmeleri nedeniyle önemli unsurlardır. RabbitMQ, birden fazla tüketicinin bir konuya abone olabileceği ve bir mesajın birden fazla kopyasını alabileceği geleneksel bir mesajlaşma sisteminin güçlü bir örneğidir.

Öte yandan, olay akışı kalıcılığa odaklanır. Olaylar arşivlenmeli, saklanmalı ve bir kez işlenmelidir. Olayların belirli tüketicilere yönlendirilmesi önemli değildir, çünkü tüm tüketicilerin olayları aynı şekilde işlemesi fikri esastır.

Apache Pulsar, Apache Yazılım Vakfı tarafından geliştirilen ve olay akışını destekleyen açık kaynaklı bir mesajlaşma sistemidir. Başlangıçta Kafka ile geliştirilen Pulsar, başlangıçta geleneksel bir mesaj kuyruğu çözümü olarak ortaya çıkmış ve daha sonra olay akışı yetenekleri edinmiştir. Bu nedenle Pulsar, ayrı uygulamalar dağıtmaya gerek kalmadan her iki yaklaşımın bir kombinasyonuna ihtiyaç duyulduğunda kullanışlıdır.

Sonuç

Artık sunucunuzun arka planında güvenli bir şekilde çalışan ve bir sistem hizmeti olarak yapılandırılmış Apache Kafka'ya sahipsiniz. Ayrıca, komut satırından konuları nasıl yöneteceğinizi, mesajları nasıl oluşturup kullanacağınızı da öğrendiniz. Ancak, Kafka'nın asıl cazibesi, uygulamalarınıza entegre edebileceğiniz geniş istemci yelpazesidir.