Data Transmission Service：Gunakan klien Kafka untuk mengonsumsi data yang dilacak

Catatan penggunaan

• Jika Anda mengaktifkan komit otomatis saat menggunakan fitur pelacakan perubahan, beberapa data mungkin dikomit sebelum dikonsumsi, yang dapat menyebabkan kehilangan data. Kami menyarankan Anda untuk secara manual mengkomit data.

  Catatan

  Jika data gagal dikomit, Anda dapat memulai ulang klien untuk melanjutkan konsumsi data dari titik pemeriksaan terakhir yang direkam. Namun, data duplikat mungkin dihasilkan selama periode ini. Anda harus secara manual menyaring data duplikat tersebut.

• Data diserialisasi dan disimpan dalam format Avro. Untuk informasi lebih lanjut, lihat Record.avsc.

  Peringatan

  Jika Anda tidak menggunakan klien Kafka yang dijelaskan dalam topik ini, Anda harus mengurai data yang dilacak berdasarkan skema Avro dan memeriksa data yang telah diurai.

• Satuan pencarian adalah detik ketika Data Transmission Service (DTS) memanggil operasi offsetForTimes. Satuan pencarian adalah milidetik ketika klien Kafka asli memanggil operasi ini.

• Koneksi transien mungkin terjadi antara klien Kafka dan server pelacakan perubahan karena beberapa alasan, seperti pemulihan bencana. Jika Anda tidak menggunakan klien Kafka yang dijelaskan dalam topik ini, klien Kafka Anda harus memiliki kemampuan penyambungan ulang jaringan.

• Jika Anda menggunakan klien Kafka asli untuk mengonsumsi data yang dilacak, modul pengumpulan data inkremental mungkin berubah di DTS. Dalam mode langganan, titik pemeriksaan konsumsi yang disimpan oleh klien Kafka ke server DTS dihapus. Anda perlu menentukan titik pemeriksaan konsumsi untuk mengonsumsi data yang dilacak sesuai dengan kebutuhan bisnis Anda. Jika Anda ingin mengonsumsi data dalam mode langganan, kami menyarankan Anda menggunakan demo SDK yang disediakan oleh DTS untuk melacak dan mengonsumsi data, atau secara manual mengelola titik pemeriksaan konsumsi. Untuk informasi lebih lanjut, lihat Gunakan Demo SDK untuk Mengonsumsi Data yang Dilacak dan bagian Kelola Titik Pemeriksaan Konsumsi dari topik ini.

Jalankan klien Kafka

Unduh demo klien Kafka. Untuk informasi lebih lanjut tentang cara menggunakan demo, lihat Readme.

Prosedur

Langkah-langkah berikut menunjukkan cara menjalankan klien Kafka untuk mengonsumsi data yang dilacak. Dalam contoh ini, IntelliJ IDEA Community Edition 2018.1.4 untuk Windows digunakan.

1. Buat instance pelacakan perubahan. Untuk informasi lebih lanjut, lihat Ikhtisar Skenario Pelacakan Perubahan.

2. Buat satu atau lebih grup konsumen. Untuk informasi lebih lanjut, lihat Buat Grup Konsumen.

3. Unduh paket demo klien Kafka dan dekompresi paket tersebut.

   Catatan

   Klik dan pilih Download ZIP untuk mengunduh paket.

4. Buka IntelliJ IDEA. Di jendela yang muncul, klik Open.

   

5. Di kotak dialog yang muncul, pergi ke direktori tempat demo yang diunduh berada. Temukan file pom.xml.

   

6. Di kotak dialog yang muncul, pilih Open as Project.

7. Di jendela Alat Proyek IntelliJ IDEA, klik folder untuk menemukan file demo klien Kafka, dan klik dua kali file tersebut. Nama file adalah NotifyDemoDB.java.

8. Tentukan parameter dalam file NotifyDemoDB.java.

   

   Parameter	Deskripsi	Cara mendapatkan nilai parameter
   USER_NAME	Nama pengguna akun grup konsumen. Peringatan Jika Anda tidak menggunakan klien Kafka yang dijelaskan dalam topik ini, Anda harus menentukan parameter ini dalam format berikut: <Username>-<ID Grup Konsumen>. Contoh: dtstest-dtsae******bpv. Jika tidak, koneksi akan gagal.	Di konsol DTS, temukan instance pelacakan perubahan yang ingin Anda kelola dan klik ID instance. Di panel navigasi di sebelah kiri, klik Consume Data. Di halaman yang muncul, Anda dapat melihat informasi tentang grup konsumen, seperti ID atau nama dan akun grup konsumen. Catatan Kata sandi akun grup konsumen ditentukan saat Anda membuat grup konsumen.
   PASSWORD_NAME	Kata sandi akun.
   SID_NAME	ID grup konsumen.
   GROUP_NAME	Nama grup konsumen. Tetapkan parameter ini ke ID grup konsumen.
   KAFKA_TOPIC	Nama topik yang dilacak dari instance pelacakan perubahan.	Di konsol DTS, temukan instance pelacakan perubahan yang ingin Anda kelola dan klik ID instance. Di halaman Basic Information, Anda dapat melihat informasi tentang topik dan jaringan.
   KAFKA_BROKER_URL_NAME	Titik akhir instance pelacakan perubahan. Catatan Jika Anda melacak perubahan data melalui jaringan internal, latensi jaringan minimal. Ini berlaku jika instance Elastic Compute Service (ECS) tempat Anda menerapkan klien Kafka berada di jaringan klasik atau di virtual private cloud (VPC) yang sama dengan instance pelacakan perubahan. Untuk memastikan stabilitas jaringan, kami menyarankan Anda untuk tidak menggunakan titik akhir publik.
   INITIAL_CHECKPOINT_NAME	Titik pemeriksaan konsumsi data yang dikonsumsi. Nilainya adalah timestamp UNIX. Contoh: 1592269238. Catatan Anda harus menyimpan titik pemeriksaan konsumsi untuk alasan berikut: Jika proses konsumsi terganggu, Anda dapat menentukan titik pemeriksaan konsumsi pada klien Kafka untuk melanjutkan konsumsi data. Ini mencegah kehilangan data. Saat Anda memulai klien Kafka, Anda dapat menentukan titik pemeriksaan konsumsi untuk mengonsumsi data berdasarkan kebutuhan bisnis Anda. Jika parameter SUBSCRIBE_MODE_NAME diatur ke subscribe, parameter INITIAL_CHECKPOINT_NAME yang Anda tentukan hanya berlaku saat Anda memulai klien Kafka untuk pertama kalinya.	Titik pemeriksaan konsumsi data yang dikonsumsi harus berada dalam rentang data instance pelacakan perubahan. Titik pemeriksaan konsumsi harus dikonversi menjadi timestamp UNIX. Catatan Anda dapat melihat rentang data instance pelacakan perubahan di kolom Data Range pada halaman Tugas Pelacakan Perubahan. Anda dapat menggunakan mesin pencari untuk mendapatkan konverter timestamp UNIX.
   USE_CONFIG_CHECKPOINT_NAME	Menentukan apakah klien dipaksa untuk mengonsumsi data dari titik pemeriksaan konsumsi yang ditentukan. Nilai default: true. Anda dapat mengatur parameter ini ke true untuk mencegah data yang diterima tetapi belum diproses hilang.	Tidak ada
   SUBSCRIBE_MODE_NAME	Menentukan apakah menjalankan dua atau lebih klien Kafka untuk grup konsumen. Jika Anda ingin menggunakan fitur ini, atur parameter ini ke subscribe untuk klien Kafka ini. Nilai default adalah assign, yang menunjukkan bahwa fitur ini tidak digunakan. Kami menyarankan Anda hanya menerapkan satu klien Kafka untuk grup konsumen.	Tidak ada

9. Di bilah menu atas IntelliJ IDEA, pilih Run > Run untuk menjalankan klien.

   Catatan

   Jika Anda menjalankan IntelliJ IDEA untuk pertama kali, waktu tertentu diperlukan untuk memuat dan menginstal dependensi yang relevan.

Hasil pada klien Kafka

Gambar berikut menunjukkan bahwa klien Kafka dapat melacak perubahan data dari database sumber.

Anda dapat menghapus dua garis miring ke depan (//) dari string //log.info(ret) di baris 25 file NotifyDemoDB.java. Kemudian, jalankan klien lagi untuk melihat informasi perubahan data.

FAQ

• T: Mengapa saya perlu mencatat titik pemeriksaan konsumsi klien Kafka?

  J: Titik pemeriksaan konsumsi yang dicatat oleh DTS adalah titik waktu ketika DTS menerima operasi commit dari klien Kafka. Titik pemeriksaan konsumsi yang dicatat mungkin berbeda dari waktu konsumsi aktual. Jika aplikasi bisnis atau klien Kafka terganggu secara tak terduga, Anda dapat menentukan titik pemeriksaan konsumsi yang akurat untuk melanjutkan konsumsi data. Ini mencegah kehilangan data atau konsumsi data duplikat.

Kelola titik pemeriksaan konsumsi

1. Konfigurasikan klien Kafka untuk mendengarkan pergantian modul pengumpulan data di DTS.

   Anda dapat mengonfigurasi properti konsumen klien Kafka untuk mendengarkan pergantian modul pengumpulan data di DTS. Kode berikut memberikan contoh tentang cara mengonfigurasi properti konsumen:

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

   Kode berikut memberikan contoh tentang cara mengimplementasikan 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. Tentukan titik pemeriksaan konsumsi berdasarkan pergantian modul pengumpulan data yang ditangkap di DTS.

   Atur titik pemeriksaan konsumsi awal pelacakan data berikutnya ke timestamp entri data terlacak terbaru yang dikonsumsi oleh klien. Kode berikut memberikan contoh tentang cara menentukan titik pemeriksaan konsumsi:

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

   Catatan

   Untuk informasi lebih lanjut tentang contoh-contoh, lihat KafkaRecordFetcher.

Pemetaan antara tipe data dan nilai dataTypeNumber

Pemetaan antara tipe data Oracle dan nilai dataTypeNumber

Pemetaan antara tipe data PostgreSQL dan nilai dataTypeNumber