DataHub Kafka uyumluluk modu
DataHub artık Apache Kafka protokolüyle uyumludur. Veri yayınlamak ve abone olmak için standart bir Kafka SDK kullanarak DataHub'a bağlanabilirsiniz.
DataHub ile Kafka arasındaki kavram eşleştirmesi
|
|
|
|
|
|
|
|
|
|
|
|
Kafka topic
DataHub'da da topic bulunur. Ancak DataHub, Kafka'da mevcut olmayan proje adında ek bir kaynak katmanı içerir. Uyumluluğu sağlamak için bir Kafka topic'i, nokta (.) ile birleştirilen bir DataHub projesi ve topic kombinasyonuna karşılık gelir. Örneğin, DataHub'da test_project adlı bir projeniz ve test_topic adlı bir topic'iniz varsa, karşılık gelen Kafka topic adı test_project.test_topic olur.
Kafka partition
Bir Kafka partition'ı doğrudan bir DataHub shard'ına eşdeğerdir. Her ikisi de sıralı bir veri kuyruğunu temsil eder.
Kafka consumer group
Bir DataHub consumer group'u, bir Kafka consumer group'u gibi davranır. Bir proje içinde bir consumer group oluşturabilir ve bunu abone olmak istediğiniz topic'lerle ilişkilendirebilirsiniz. Grup, o proje içindeki birden fazla topic'e abone olabilir. Bir consumer group bir topic'e bağlandığında, abonelik otomatik olarak oluşturulur ve topic'in abonelik listesi sayfasında görünür. Bu aboneliği silmek, grubun topic'e abone olmasını engeller ve önceki tüm consumer offset'lerini siler.
Consumer group bir projenin alt kaynağı olduğundan, bunu projeyle birlikte belirtmeniz gerekir. Örneğin, DataHub projeniz test_project ve consumer group'unuz test_group ise, Kafka grup adı test_project.test_group olur.
Consumer group oluştur iletişim kutusunda, bir Name ve Description girin, Topic alanında abone olunacak topic'leri seçin, sağdaki seçili listeye eklemek için > düğmesini tıklayın ve ardından Create öğesini tıklayın.
Her grup en fazla 50 topic'e abone olabilir. Daha fazlasına abone olmanız gerekiyorsa lütfen bir destek talebi gönderin.
Kafka Record
Bir Kafka kaydı anahtar-değer biçimini kullanır. DataHub kayıtları iki biçimde mevcuttur: Tuple (güçlü yapılandırılmış veri) ve Blob (ikili veri). Bir Kafka kaydı iki bölümden oluşur: bir başlık ve bir anahtar-değer çifti.
Kafka Header
Bir Kafka başlığı, kayıtlara meta veri ekler ve bu bir DataHub niteliğine eşdeğerdir. Başlıklarla veri yazmak için bir Kafka istemcisi kullandığınızda, başlık bilgileri DataHub nitelikleri olarak depolanır. Bir Kafka başlığının değeri NULL ise, ilgili başlık yoksayılır. Rezerve edilmiş bir iç anahtar olduğu için "__kafka_key__" ifadesini başlık anahtarı olarak kullanmayın.
Kafka anahtar-değer verileri
Bir DataHub topic'i Tuple türündeyse, anahtarı ilk String sütuna ve değeri ikinci String sütuna karşılık gelir.
Bir DataHub topic'i Blob türündeyse, değeri veri içeriğine karşılık gelir ve anahtarı <"__kafka_key__", key> biçimindeki bir niteliğe yerleştirilir.
Kafka offset
Bir Kafka offset'i, 0'dan başlayan ve her partition içinde otomatik olarak artan 64 bitlik bir tamsayıdır; bu, her kayda benzersiz bir konum verir. Temel olarak consumer ilerlemesini izlemek için kullanılır. Bir DataHub sequence'i doğrudan bir Kafka offset'ine eşdeğerdir.
Sınırlamalar
DataHub, Kafka işlemlerini, idempotency özelliğini, SchemaRegistry'yi veya Log Compaction'ı desteklemez.
Hızlı Başlangıç
Önerilen Kafka istemci sürümü 2.4.0'ı kullanın. Bu sürüm 0.10.0 ile 4.0 arasındaki sürümlerle uyumludur.
DataHub'da ilgili Project, Topic ve Group kaynaklarını oluşturun.
Güvenlik kimlik doğrulama yöntemini SASL/SSL olarak değiştirin, SASL için PLAIN mekanizmasını kullanın ve Alibaba Cloud AccessKey çiftinizi (AK/SK) yapılandırın.
Kafka broker bilgilerini DataHub uç noktasına değiştirin (hizmet alanları listesine bakın).
Kaynak oluşturma
İlk olarak, DataHub konsoluna giriş yapın ve bir proje oluşturun.
Yeni Proje iletişim kutusunda, Name değerini test_kafka_project ve Description değerini test kafka olarak ayarlayın.
Topic oluşturma
DataHub konsolunun New Topic sayfasında, oluşturma yöntemi olarak Direct Create öğesini seçin, ad olarak test_kafka girin ve tür olarak TUPLE seçeneğini belirleyin. Şema ayrıntıları bölümünde, key ve value adlı iki alan ekleyin, her ikisinin türünü STRING olarak ayarlayın ve Allow NULL seçeneğini işaretleyin. Shard Count değerini 1 ve Lifecycle değerini 3 olarak ayarlayın, Shard Scaling Mode özelliğini etkinleştirin ve Multi-version özelliğini devre dışı bırakın. Bir açıklama girin ve Create öğesini tıklayın.
Bir grup oluşturun ve tüketmeniz gereken topic'leri bağlayın. Grup oluşturulduktan sonra bağlı topic listesini değiştirebilirsiniz. Mesajları tüketmeniz gerekmiyorsa bu adımı atlayın.
New Group panelinde, bir Name ve Description girin, transfer listesini kullanarak soldaki kullanılabilir listeden sağdaki seçili listeye tüketilecek topic'leri taşıyın ve ardından Create öğesini tıklayın.
Kimlik Doğrulama Yöntemi
kafka_client_producer_jaas.conf adlı bir dosya oluşturun ve aşağıdaki içerikle herhangi bir yola kaydedin:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="accessId"
password="accessKey";
};kafka_client_producer_jaas.conf adlı bir dosya oluşturun ve aşağıdaki içerikle herhangi bir dizine kaydedin:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="YOUR_ACCESS_ID"
password="YOUR_ACCESS_KEY";
};Üretici örneği
kafka clients pom dosyası
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.0</version>
</dependency>
Kafka istemci pom.xml dosyası:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.0</version>
</dependency>
ProducerExample
Tüketici örneği
ConsumerExample
package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample2 {
static {
System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
}
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "PLAIN");
properties.put("group.id", "test_project.test_kafka_group");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "60000");
properties.put("heartbeat.interval.ms", "40000");
properties.put("ssl.endpoint.identification.algorithm", "");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = new ArrayList<>();
topicList.add("test_project.test_topic1");
topicList.add("test_project.test_topic2");
topicList.add("test_project.test_topic3");
kafkaConsumer.subscribe(topicList);
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
for (ConsumerRecord<String, String> record : records) {
System.out.println(record.toString());
}
}
}
}ConsumerExample
package com.aliyun.datahub.kafka.demo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class ConsumerExample {
static {
System.setProperty("java.security.auth.login.config", "/path/to/your/kafka_client_producer_jaas.conf");
}
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "PLAIN");
properties.put("group.id", "test_project.test_kafka_group");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "60000");
properties.put("heartbeat.interval.ms", "40000");
properties.put("ssl.endpoint.identification.algorithm", "");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = new ArrayList<>();
topicList.add("test_project.test_topic1");
topicList.add("test_project.test_topic2");
topicList.add("test_project.test_topic3");
kafkaConsumer.subscribe(topicList);
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
for (ConsumerRecord<String, String> record : records) {
System.out.println(record.toString());
}
}
}
}Streams Örnekleri
Bu kod, test_project içindeki input konusundan veri okur, anahtar ve değer dizelerini küçük harfe dönüştürür ve veriyi output konusuna yeniden yazar.
public class StreamExample {
static {
System.setProperty("java.security.auth.login.config", "/path/xxx/kafka_client_producer_jaas.conf");
}
public static void main(final String[] args) {
final String input = "test_project.input";
final String output = "test_project.output";
final Properties properties = new Properties();
properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
properties.put("application.id", "test_project.test_kafka_group");
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "PLAIN");
properties.put("session.timeout.ms", "60000");
properties.put("heartbeat.interval.ms", "40000");
properties.put("auto.offset.reset", "earliest");
final StreamsBuilder builder = new StreamsBuilder();
TestMapper testMapper = new TestMapper();
builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
.map(testMapper)
.to(output, Produced.with(Serdes.String(), Serdes.String()));
final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
final CountDownLatch latch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
@Override
public void run() {
streams.close();
latch.countDown();
}
});
try {
streams.start();
latch.await();
} catch (final Throwable e) {
System.exit(1);
}
System.exit(0);
}
static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
@Override
public KeyValue<String, String> apply(String s, String s2) {
return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
}
}
}Bu örnek, test_project projesindeki input topic'inden veri okur, anahtar ve değer dizelerini küçük harfe dönüştürür ve sonucu output topic'ine yazar.
public class StreamExample {
static {
System.setProperty("java.security.auth.login.config", "/path/to/your/kafka_client_producer_jaas.conf");
}
public static void main(final String[] args) {
final String input = "test_project.input";
final String output = "test_project.output";
final Properties properties = new Properties();
properties.put("bootstrap.servers", "dh-cn-hangzhou.aliyuncs.com:9092");
properties.put("application.id", "test_project.test_kafka_group");
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "PLAIN");
properties.put("session.timeout.ms", "60000");
properties.put("heartbeat.interval.ms", "40000");
properties.put("auto.offset.reset", "earliest");
final StreamsBuilder builder = new StreamsBuilder();
TestMapper testMapper = new TestMapper();
builder.stream(input, Consumed.with(Serdes.String(), Serdes.String()))
.map(testMapper)
.to(output, Produced.with(Serdes.String(), Serdes.String()));
final KafkaStreams streams = new KafkaStreams(builder.build(), properties);
final CountDownLatch latch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
@Override
public void run() {
streams.close();
latch.countDown();
}
});
try {
streams.start();
latch.await();
} catch (final Throwable e) {
System.exit(1);
}
System.exit(0);
}
static class TestMapper implements KeyValueMapper<String, String, KeyValue<String, String>> {
@Override
public KeyValue<String, String> apply(String s, String s2) {
return new KeyValue<>(StringUtils.lowerCase(s), StringUtils.lowerCase(s2));
}
}
}Streams işini başlattıktan sonra, shard ataması yaklaşık bir dakika sürer. Ardından konsolda görev sayısını görüntüleyebilirsiniz. Görev sayısı, input topic'indeki shard sayısıyla eşleşir. Bu örnekte, input topic'i üç shard'a sahiptir.
Çıkış örnekleme sonucu, verilerin doğru şekilde yazıldığını gösterir. Streams görevindeki TestMapper tarafından işlendikten sonra, büyük harfli girdi küçük harfli çıktıya dönüştürüldü: anahtarlar aaaa, cccc ve eeee, karşılık gelen değerler ise bbbb, dddd ve ffff şeklindedir.
Self-managed Kafka'dan DataHub'a veri geçirin
Broker adresini değiştirin. Daha fazla bilgi için aşağıdaki alan adı listesine bakın.
DataHub'da kaynaklar oluşturun ve kodunuzdaki kaynak adlarını değiştirin. Daha fazla bilgi için yukarıdaki kaynak oluşturma bölümüne bakın.
Kimlik doğrulama yöntemini SASL_SSL olarak değiştirin. Kimlik doğrulama mekanizması PLAIN'dir. Kullanıcı adınızı Alibaba Cloud AccessKey ID'niz (AK) ve şifrenizi AccessKey Secret'ınız (SK) olarak ayarlayın.
Ek Bilgiler
Yapılandırma Genel Bakışı
C=Tüketici, P=Üretici, S=Streams
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Hizmet Alan Adı Listesi
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Kafka API uyumluluğu
Resmi Kafka belgeleri mevcut tüm API'leri listeler. Aşağıdaki tabloda, DataHub'un Kafka uyumluluk modunda desteklediği API'ler listelenmiştir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|