DataWorks：Sinkronisasi offline tabel tunggal dari Kafka ke MaxCompute

Langkah 1: Buat node dan konfigurasikan tugas

Untuk langkah umum membuat dan mengonfigurasi node di Antarmuka tanpa kode, lihat panduan Codeless UI configuration. Topik ini tidak mengulangi langkah-langkah tersebut.

Langkah 2: Konfigurasikan sumber data dan tujuan

Langkah 3: Konfigurasikan pemetaan bidang

Setelah Anda memilih sumber data dan tujuan, Anda harus menentukan pemetaan kolom antara reader dan writer. Anda dapat memilih Map Fields with Same Name, Map Fields in Same Line, Clear Mappings, or Manually Edit Mapping.

• Terdapat enam bidang default di sisi Kafka.

  Field Name	Description
  __key__	Kunci record Kafka.
  __value__	Nilai record Kafka.
  __partition__	Nomor partisi tempat record Kafka berada. Nomor partisi dimulai dari 0.
  __headers__	Header record Kafka.
  __offset__	Offset record Kafka dalam partisinya. Offset dimulai dari 0.
  __timestamp__	Timestamp milidetik integer 13 digit dari record Kafka.

• Anda dapat mengonfigurasi penguraian JSON kustom untuk bidang di sisi Kafka. Gunakan sintaks . (untuk mengambil subfield) dan [] (untuk mengambil elemen array) untuk mendapatkan konten dari bidang nilai record Kafka berformat JSON.

  Penting

  Jika nama bidang JSON mengandung karakter ".", Anda tidak dapat mengambil nilai bidang tersebut dengan mendefinisikan bidang tersebut. Hal ini karena menyebabkan ambiguitas dalam sintaks definisi bidang.

  Berikut adalah contoh nilai data record berformat JSON di Kafka.

  {
        "a": {
        "a1": "hello"
        },
        "b": "world",
        "c":[
              "xxxxxxx",
              "yyyyyyy"
              ],
        "d":[
              {
                    "AA":"this",
                    "BB":"is_data"
              },
              {
                    "AA":"that",
                    "BB":"is_also_data"
              }
          ],
       "a.b": "unreachable"
  }

  • Untuk menyinkronkan data a1, "hello", Anda dapat menambahkan bidang a.a1 di sisi Kafka.

  • Untuk menyinkronkan data b, "world", Anda dapat menambahkan bidang b di sisi Kafka.

  • Untuk menyinkronkan data c, "yyyyyyy", Anda dapat menambahkan bidang c[1] di sisi Kafka.

  • Untuk menyinkronkan data AA, "this", Anda dapat menambahkan bidang d[0].AA di sisi Kafka.

  • Jika bidang di sisi Kafka didefinisikan sebagai a.b, Anda tidak dapat menyinkronkan data "unreachable".

• Bidang tabel sumber atau tujuan dapat dikecualikan dari pemetaan. Instans sinkronisasi tidak membaca bidang sumber yang tidak dipetakan. NULL ditulis ke bidang tujuan yang tidak dipetakan.

• Anda tidak dapat memetakan satu bidang sumber ke beberapa bidang tujuan. Bidang tujuan juga tidak dapat dipetakan dari beberapa bidang sumber.

Langkah 4: Konfigurasikan parameter lanjutan

Klik Advanced Configuration di sisi kanan tugas untuk mengatur parameter seperti Maximum Expected Concurrency dan Policy for Dirty Data Records. Untuk tutorial ini, atur Policy for Dirty Data Records ke Ignore Dirty Data Records dan biarkan parameter lain pada nilai default-nya. Untuk informasi selengkapnya, lihat Codeless UI Configuration.

Langkah 5: Konfigurasikan dan jalankan pengujian

1. Di sisi kanan halaman pengeditan node sinkronisasi offline, klik Debugging Configurations. Atur Script Parameters dan Resource Group untuk pengujian. Lalu, klik Run di bilah alat atas untuk menguji apakah tautan sinkronisasi berjalan sukses.

2. Anda dapat mengklik ikon di panel navigasi kiri lalu klik ikon di sebelah kanan Personal Directory untuk membuat file dengan ekstensi .sql. Anda kemudian dapat menjalankan kueri SQL berikut untuk memeriksa apakah data di tabel tujuan sesuai harapan.

   Catatan

   • Untuk mengkueri data dengan cara ini, Anda harus attach proyek MaxCompute tujuan ke DataWorks sebagai resource komputasi.

   • Di halaman pengeditan file .sql, klik Debugging Configurations di sisi kanan. Tentukan Type sumber data, Computing Resource, dan Resource Group. Lalu, klik Run di bilah alat atas.

   SELECT * FROM <MaxCompute_destination_table_name> WHERE pt=<specified_partition> LIMIT 20;

Langkah 6: Konfigurasikan penjadwalan dan publikasikan tugas

Klik Scheduling di sisi kanan tugas sinkronisasi offline. Atur parameter konfigurasi penjadwalan untuk eksekusi berulang. Lalu, klik Publish di bilah alat atas untuk membuka panel penerbitan. Ikuti petunjuk di layar untuk publish tugas.

Saat Anda mengonfigurasi sumber data dan tujuan, Anda menggunakan tiga parameter penjadwalan: ${startTime}, ${endTime}, dan ${partition}. Dalam konfigurasi penjadwalan, Anda harus menentukan kebijakan penggantian untuk parameter-parameter ini berdasarkan kebutuhan sinkronisasi Anda. Berikut adalah contoh konfigurasi untuk beberapa skenario khas.

Atur siklus penjadwalan berdasarkan interval yang diinginkan.