Data Transmission Service:Konsumsi data langganan dengan Kafka

Catatan

• Saat menggunakan demo dalam topik ini, auto commit dapat menyebabkan kehilangan data karena offset mungkin dikomit sebelum pemrosesan data selesai. Untuk mencegah hal ini, kami merekomendasikan commit manual.

  Catatan

  Jika commit gagal, klien akan memulai ulang konsumsi dari titik pemeriksaan konsumsi terakhir yang terekam. Hal ini dapat menyebabkan duplikasi data, yang harus Anda tangani dalam logika aplikasi Anda.

• Data disimpan menggunakan serialisasi Avro. Untuk detail format, lihat file Record.avsc.

  Peringatan

  Jika Anda menggunakan klien Kafka selain yang disediakan dalam topik ini, data mungkin diurai secara salah selama deserialisasi (Contoh deserialisasi Avro DTS). Anda harus memverifikasi data tersebut.

• Di DTS, API offsetForTimes menggunakan satuan detik, sedangkan API Kafka asli menggunakan milidetik.

• Server Data Subscription mungkin mengalami gangguan jaringan sementara akibat peristiwa seperti disaster recovery. Jika Anda tidak menggunakan klien Kafka yang disediakan dalam topik ini, pastikan klien Anda mendukung retry jaringan.

• Jika Anda menggunakan klien Kafka asli dalam mode subscribe, titik pemeriksaan konsumsi yang disimpan di server mungkin dihapus saat DTS mengganti modul pengumpulan data inkrementalnya. Dalam kasus ini, Anda harus menyesuaikan titik pemeriksaan konsumsi secara manual. Jika Anda perlu menggunakan mode subscribe, kami merekomendasikan penggunaan SDK yang disediakan DTS untuk mengonsumsi data Data Subscription dan mengelola titik pemeriksaan konsumsi sendiri. Untuk informasi lebih lanjut, lihat Konsumsi data Data Subscription menggunakan SDK dan Kelola titik pemeriksaan konsumsi.

Waktu proses klien Kafka

Unduh demo klien Kafka. Untuk petunjuk penggunaan, lihat file Readme pada demo tersebut.

Prosedur

Jalankan demo klien Kafka di IntelliJ IDEA untuk mengonsumsi data dari saluran langganan data.

1. Buat saluran langganan data. Untuk informasi lebih lanjut, lihat Ikhtisar solusi langganan.

2. Buat satu atau beberapa kelompok konsumen. Untuk informasi lebih lanjut, lihat Buat kelompok konsumen.

3. Unduh demo klien Kafka dan ekstrak file tersebut.

   Catatan

   Klik , lalu pilih Download ZIP untuk mengunduh file.

4. Buka IntelliJ IDEA dan klik Open.

5. Pada kotak dialog, arahkan ke direktori tempat Anda mengekstrak demo klien Kafka. Perluas folder seperti yang ditunjukkan pada gambar berikut untuk menemukan file Project Object Model (POM): pom.xml.

6. Pada kotak dialog yang muncul, klik Open as Project.

7. Di jendela IntelliJ IDEA, perluas folder, lalu temukan dan klik ganda file demo klien Kafka: NotifyDemoDB.java.

8. Atur nilai parameter dalam file NotifyDemoDB.java.

   Parameter	Deskripsi	Cara mendapatkan
   USER_NAME	Username untuk kelompok konsumen. Peringatan Jika Anda menggunakan klien berbeda, Anda harus mengatur username dalam format <akun kelompok konsumen>-<ID kelompok konsumen>. Contoh: dtstest-dtsae******bpv.	Di Konsol DTS, klik ID instans langganan target. Di panel navigasi kiri, klik Consume Data. Pada halaman yang muncul, Anda dapat menemukan informasi Consumer Group ID/Name dan Account untuk kelompok konsumen. Catatan Password untuk akun kelompok konsumen ditentukan saat Anda membuat kelompok konsumen.
   PASSWORD_NAME	Password untuk akun kelompok konsumen.
   SID_NAME	ID kelompok konsumen.
   GROUP_NAME	Nilai ini harus sesuai dengan ID kelompok konsumen.
   KAFKA_TOPIC	Topik langganan untuk saluran langganan data.	Di Konsol DTS, klik ID instans langganan target. Pada halaman Basic Information, Anda dapat menemukan informasi Topic dan Network.
   KAFKA_BROKER_URL_NAME	Alamat jaringan untuk saluran langganan data. Catatan Jika klien Kafka dan saluran langganan data Anda berada dalam jaringan klasik atau VPC yang sama, gunakan alamat IP internal untuk meminimalkan latensi jaringan. Kami tidak merekomendasikan penggunaan titik akhir publik karena potensi ketidakstabilan jaringan.
   INITIAL_CHECKPOINT_NAME	Waktu mulai konsumsi data, ditentukan sebagai Stempel waktu UNIX. Contoh: 1592269238. Catatan Menyimpan stempel waktu memungkinkan Anda: Melanjutkan konsumsi dari stempel waktu terakhir yang disimpan setelah terjadi gangguan untuk mencegah kehilangan data. Memulai konsumsi data dari titik waktu tertentu. Jika SUBSCRIBE_MODE_NAME diatur ke subscribe, nilai INITIAL_CHECKPOINT_NAME hanya berlaku saat klien langganan pertama kali dijalankan.	Waktu mulai konsumsi harus berada dalam rentang data instans langganan dan harus dikonversi ke Stempel waktu UNIX. Catatan Anda dapat melihat rentang data instans langganan di kolom Data Range pada daftar tugas langganan. Anda dapat menggunakan mesin pencari untuk menemukan konverter Stempel waktu UNIX.
   USE_CONFIG_CHECKPOINT_NAME	Nilai default adalah true, yang memaksa klien untuk memulai dari waktu yang ditentukan. Hal ini mencegah kehilangan data yang telah diterima tetapi belum diproses.	Tidak ada
   SUBSCRIBE_MODE_NAME	Untuk menjalankan dua atau lebih klien Kafka dalam kelompok konsumen yang sama, atur parameter ini ke subscribe pada semua klien. Nilai default adalah assign. Dalam mode ini, fitur ini dinonaktifkan, dan kami merekomendasikan hanya menerapkan satu klien.	Tidak ada

9. Dari bilah menu atas, pilih Run > Run untuk menjalankan klien.

   Catatan

   Saat menjalankan aplikasi untuk pertama kalinya, IntelliJ IDEA mungkin memerlukan waktu untuk mengunduh dan menginstal paket dependensi yang diperlukan secara otomatis.

Hasil

Klien berhasil berlangganan data perubahan dari database sumber.

Untuk melihat data perubahan secara detail, hapus // dari //log.info(ret); pada baris 25 dalam file NotifyDemoDB.java dan jalankan kembali klien.

FAQ

• T: Mengapa saya perlu melacak titik pemeriksaan konsumsi klien saya secara manual?

  J: Titik pemeriksaan konsumsi yang direkam oleh DTS menunjukkan kapan DTS menerima commit dari klien Kafka, bukan kapan aplikasi Anda selesai memproses data. Jika aplikasi atau klien Kafka Anda berhenti secara tak terduga, mencatat titik pemeriksaan sendiri memungkinkan Anda melanjutkan konsumsi dari titik gangguan yang tepat, sehingga mencegah duplikasi data maupun kehilangan data.

Kelola checkpoint

1. Konfigurasikan klien langganan untuk mendengarkan pergantian kluster dalam modul pengumpulan data DTS.

   Untuk melakukan ini, daftarkan ClusterSwitchListener sebagai interceptor konsumen dengan mengatur properti berikut untuk consumer:

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   Kode berikut menunjukkan contoh implementasi ClusterSwitchListener:

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. Tangani event pergantian kluster dari modul pengumpulan data DTS.

   Saat ClusterSwitchException tertangkap, atur ulang checkpoint awal untuk langganan berikutnya ke stempel waktu dari catatan terakhir yang berhasil dikonsumsi. Potongan kode berikut memberikan contoh:

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   Catatan

   Untuk contoh implementasi, lihat KafkaRecordFetcher.

Pemetaan tipe data dan dataTypeNumber

Konsumsi data langganan menggunakan Python

Jika aplikasi konsumsi Anda ditulis dalam Python, Anda dapat menggunakan pustaka kafka-python dan avro untuk mengonsumsi dan mendeserialisasi data berformat Avro dari saluran langganan data. Contoh berikut telah diverifikasi dengan kafka-python 3.0.7 dan avro 1.12.1.

1. Instal pustaka Python yang diperlukan.

   pip install kafka-python avro

2. Unduh file skema Avro Record.avsc dari direktori avro repositori demo klien Kafka. Ini adalah file skema yang sama yang dirujuk dalam bagian Catatan topik ini.

3. Gunakan kode contoh berikut untuk membuat KafkaConsumer, memuat skema Avro, dan mendeserialisasi data perubahan:

   from kafka import KafkaConsumer
   import avro.schema
   from avro.io import DatumReader, BinaryDecoder
   
   schema = avro.schema.parse(open("Record.avsc", "rb").read())
   record_schema = schema.schemas[13]
   
   consumer = KafkaConsumer(
       "<topic>",
       bootstrap_servers="<titik akses DTS:port>",
       group_id="<ID kelompok konsumen>",
       enable_auto_commit=False,
       security_protocol="SASL_SSL",
       sasl_mechanism="PLAIN",
       sasl_plain_username="<akun kelompok konsumen>",
       sasl_plain_password="<password>",
       ssl_ca_location="ca.pem",
   )
   
   for msg in consumer:
       reader = DatumReader(record_schema)
       record = reader.read(BinaryDecoder(msg.value))
       if record.get("operation") == "HEARTBEAT":
           continue
       print(record)

   Kode contoh melakukan hal berikut:

   • Mengonfigurasi KafkaConsumer untuk terhubung menggunakan transport SASL_SSL dengan autentikasi PLAIN.

   • Memuat Record.avsc dan memilih definisi skema Record (schemas[13]) untuk mendeserialisasi setiap pesan dengan DatumReader dan BinaryDecoder.

   • Menyaring pesan HEARTBEAT, yaitu pesan keepalive yang tidak berisi data perubahan.

4. Ganti placeholder berikut dengan nilai aktual Anda. Anda dapat memperoleh nilai-nilai ini dengan metode yang sama seperti yang dijelaskan dalam tabel parameter bagian Prosedur (KAFKA_TOPIC, KAFKA_BROKER_URL_NAME, SID_NAME, USER_NAME, dan PASSWORD_NAME):

   Placeholder	Deskripsi
   <topic>	Topik langganan untuk saluran langganan data. Sesuai dengan KAFKA_TOPIC.
   <titik akses DTS:port>	Alamat jaringan saluran langganan data. Sesuai dengan KAFKA_BROKER_URL_NAME.
   <ID kelompok konsumen>	ID kelompok konsumen. Sesuai dengan SID_NAME.
   <akun kelompok konsumen>	Username untuk kelompok konsumen. Sesuai dengan USER_NAME.
   <password>	Password untuk akun kelompok konsumen. Sesuai dengan PASSWORD_NAME.