Realtime Compute for Apache Flink:Ingesti log waktu nyata

Prasyarat

• Pastikan pengguna RAM atau peran RAM Anda memiliki izin yang diperlukan untuk mengakses konsol Realtime Compute for Apache Flink. Untuk informasi selengkapnya, lihat Izin.

• Anda telah membuat ruang kerja Flink. Untuk informasi selengkapnya, lihat Buat ruang kerja.

• Penyimpanan hulu dan hilir

  • Anda telah membuat instans ApsaraMQ for Kafka. Untuk informasi selengkapnya, lihat Langkah 2: Beli dan terapkan instans.

  • Anda telah membuat instans Hologres. Untuk informasi selengkapnya, lihat Beli instans Hologres.

  Catatan

  Instans ApsaraMQ for Kafka dan Hologres harus berada di Wilayah dan VPC yang sama dengan ruang kerja Realtime Compute for Apache Flink Anda. Jika tidak, Anda harus menetapkan konektivitas jaringan. Untuk informasi selengkapnya, lihat Akses layanan lintas VPC atau Akses internet.

Langkah 1: Konfigurasikan daftar putih IP

Untuk mengizinkan Flink mengakses instans Kafka dan Hologres Anda, tambahkan Blok CIDR ruang kerja Flink ke daftar putih IP Kafka dan Hologres.

1. Dapatkan Blok CIDR VPC ruang kerja Flink Anda.

   1. Masuk ke konsol Realtime Compute for Apache Flink.

   2. Pada kolom Actions dari workspace target, pilih More > Workspace Details.

   3. Pada kotak dialog Workspace Details, temukan CIDR block VSwitch.

      Kotak dialog ini menampilkan informasi dasar ruang kerja dan daftar VSwitch. Pada kolom CIDR block, lihat blok CIDR untuk setiap zona ketersediaan. Catat blok CIDR ini untuk langkah berikutnya.

2. Tambahkan Blok CIDR ruang kerja Flink ke daftar putih IP instans Kafka Anda.

   Anda harus mengonfigurasi daftar putih IP untuk titik akhir VPC. Untuk langkah-langkah selengkapnya, lihat Konfigurasi daftar putih IP. Di kotak dialog pengeditan daftar putih, klik Add Whitelist IP untuk menambahkan Blok CIDR.

3. Tambahkan Blok CIDR ruang kerja Flink ke daftar putih IP instans Hologres Anda.

   Masuk ke instans Hologres Anda dan konfigurasikan daftar putih IP-nya. Untuk langkah-langkah selengkapnya, lihat daftar putih IP. Pada halaman konfigurasi daftar putih di Pusat Keamanan HoloWeb, masukkan Blok CIDR pada bidang IP Address dalam kotak dialog Edit IP Whitelist, lalu klik OK.

Langkah 2: Siapkan data uji Kafka

Gunakan konektor Faker Realtime Compute for Apache Flink untuk menghasilkan dan menulis data ke ApsaraMQ for Kafka. Ikuti langkah-langkah berikut di konsol pengembangan Realtime Compute for Apache Flink.

1. Di konsol ApsaraMQ for Kafka, buat topik bernama users.

   Untuk informasi selengkapnya, lihat Buat topik.

2. Buat pekerjaan untuk menulis data ke ApsaraMQ for Kafka.

   1. Masuk ke konsol pengembangan Realtime Compute for Apache Flink.

   2. Pada kolom Actions ruang kerja target, klik Console.

   3. Di panel navigasi sebelah kiri, pilih Development > ETL.

   4. Klik ikon , lalu klik New Stream Draft. Masukkan File Name dan pilih Engine Version.

      Realtime Compute for Apache Flink juga menyediakan berbagai templat kode dan sinkronisasi data. Setiap templat mencakup kasus penggunaan spesifik, contoh kode, dan instruksi. Klik templat untuk mempelajari fitur dan sintaks Realtime Compute for Apache Flink secara cepat serta menerapkan logika bisnis Anda. Untuk informasi selengkapnya, lihat Templat kode dan Templat sinkronisasi data.

      Parameter	Deskripsi	Contoh
      File Name	Nama pekerjaan. Catatan Nama pekerjaan harus unik dalam ruang kerja saat ini.	flink-test
      Engine Version	Versi engine Flink untuk pekerjaan saat ini. Pilih versi yang diberi label Recommended atau Stable untuk keandalan dan kinerja yang lebih tinggi. Untuk informasi selengkapnya tentang versi engine, lihat Catatan rilis dan Versi engine.	vvr-8.0.8-flink-1.17

   5. Klik Create.

   6. Tulis pernyataan SQL untuk pekerjaan tersebut.

      Salin kode berikut ke editor dan modifikasi parameter sesuai lingkungan Anda.

      CREATE TEMPORARY TABLE source (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        event_time TIMESTAMP
      ) WITH (
        'connector' = 'faker',
        'number-of-rows' = '100',
        'rows-per-second' = '10',
        'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}',
        'fields.first_name.expression' = '#{name.firstName}',
        'fields.last_name.expression' = '#{name.lastName}',
        'fields.address.country.expression' = '#{Address.country}',
        'fields.address.state.expression' = '#{Address.state}',
        'fields.address.city.expression' = '#{Address.city}',
        'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}'
      );
      
      CREATE TEMPORARY TABLE sink (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        `timestamp` TIMESTAMP METADATA
      ) WITH (
        'connector' = 'kafka',
        'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092',
        'topic' = 'users',
        'format' = 'json',
        'properties.enable.idempotence'='false'
      );
      
      INSERT INTO sink SELECT * FROM source;

      Tabel berikut menjelaskan parameter yang perlu Anda modifikasi.

      Parameter	Nilai contoh	Deskripsi
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	Titik akhir broker ApsaraMQ for Kafka. Daftar titik akhir broker yang dipisahkan koma dalam format 'host:port'. Anda dapat menemukan titik akhir Domain Name VPC di bagian Endpoint Information pada halaman Instance Details.
      topic	users	Nama topik ApsaraMQ for Kafka.

3. Jalankan pekerjaan.

   1. Pada halaman Development > ETL, klik Deploy.

   2. Pada kotak dialog Deploy draft, klik Confirm.

   3. Konfigurasikan sumber daya untuk pekerjaan. Untuk informasi selengkapnya, lihat Konfigurasikan sumber daya untuk pekerjaan.

   4. Pada halaman O&M > Deployments, temukan penyebaran target, lalu klik Start pada kolom Actions. Untuk informasi selengkapnya mengenai konfigurasi startup, lihat Mulai penyebaran.

   5. Pada halaman Deployments, Anda dapat memantau informasi waktu proses dan status penyebaran.

      Karena sumber Faker menghasilkan aliran terbatas, status penyebaran berubah menjadi FINISHED sekitar satu menit setelah dimulai. Saat penyebaran selesai, data telah ditulis ke topik 'users'. Kode berikut menunjukkan contoh data berformat JSON.

      {
        "id": 765,
        "first_name": "Barry",
        "last_name": "Pollich",
        "address": {
          "country": "United Arab Emirates",
          "state": "Nevada",
          "city": "Powlowskifurt"
        }
      }

Langkah 3: Buat dan jalankan pekerjaan sinkronisasi data

Langkah 4: Lihat hasil sinkronisasi lengkap

1. Masuk ke Konsol Manajemen Hologres.

2. Pada halaman Instances, klik nama instans target Anda.

3. Di pojok kanan atas halaman, klik Connect to Instance.

4. Pada tab Metadata Management, lihat skema tabel dan data tabel users yang telah disinkronkan di database flink_test_db.

   Di pohon navigasi sebelah kiri, perluas nama instans Anda > flink_test_db > test_schema > Tables. Tabel users yang telah disinkronkan akan muncul.

   Skema dan data tabel yang telah disinkronkan adalah sebagai berikut.

   • Skema tabel

     Klik ganda nama tabel users untuk melihat skema tabel.

     Skema tabel users berisi bidang-bidang berikut: id (BIGINT, primary key), first_name (TEXT), last_name (TEXT), address.country (TEXT), address.state (TEXT), dan address.city (TEXT).

     Catatan

     Selama sinkronisasi data lengkap, kami menyarankan mendefinisikan partisi dan offset metadata Kafka sebagai primary key untuk tabel Hologres. Hal ini mencegah duplikasi data jika pekerjaan gagal dan mentransmisikan ulang data.

   • Data tabel

     Di pojok kanan atas halaman tabel users, klik Query table. Masukkan perintah berikut dan klik Run.

     SELECT * FROM test_schema.users;

     Perintah tersebut mengembalikan hasil berikut.

     Kueri mengembalikan beberapa baris. Hal ini mengonfirmasi bahwa catatan berhasil disinkronkan ke tabel users. Setiap baris yang dikembalikan berisi data lengkap untuk kolom id, first_name, last_name, address.country, address.state, dan address.city.

Langkah 5: Amati sinkronisasi skema otomatis

1. Di konsol ApsaraMQ for Kafka, kirim pesan secara manual yang berisi kolom baru.

   1. Masuk ke ApsaraMQ for Kafka console.

   2. Pada halaman Instances, klik nama instans target.

   3. Pada halaman Topics, klik nama topik target (users).

   4. Klik Send Message.

   5. Konfigurasikan pesan.

      Pada kotak dialog Start to Send and Consume Message, konfigurasikan parameter sebagai berikut.

      Parameter	Contoh
      Method of Sending	Pilih Console.
      Message Key	Masukkan flinktest.
      Message Content	Salin dan tempel konten JSON berikut ke bidang Message Content. { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } } Catatan Dalam contoh ini, house-points adalah kolom bersarang baru.
      Send to Specified Partition	Pilih Yes.
      Partition ID	Masukkan 0.

   6. Klik OK.

2. Di konsol Hologres, lihat perubahan skema dan data pada tabel users.

   1. Masuk ke konsol Hologres.

   2. Pada halaman Instances, klik nama instans target.

   3. Di pojok kanan atas halaman, klik Connect to Instance.

   4. Pada tab Metadata Management, klik ganda nama tabel users.

   5. Klik Query table, masukkan pernyataan berikut, lalu klik Running.

      SELECT * FROM test_schema.users;

   6. Lihat hasil kueri.

      Kueri mengembalikan hasil berikut.

      Hasil menunjukkan bahwa catatan dengan ID 100001 berhasil ditulis ke Hologres, dan dua kolom baru, house-points.house dan house-points.points, ditambahkan ke tabel Hologres.

      Catatan

      Pesan yang dikirim ke ApsaraMQ for Kafka hanya berisi satu kolom bersarang: house-points. Namun, karena json.infer-schema.flatten-nested-columns.enable ditentukan dalam klausa WITH, Realtime Compute for Apache Flink secara otomatis meratakan kolom ini, menggunakan jalur akses bidang bersarang sebagai nama kolom baru.

Referensi

• Untuk informasi tentang pernyataan CREATE TABLE AS (CTAS), lihat Pernyataan CREATE TABLE AS (CTAS).

• Untuk informasi tentang penggunaan Message Queue for Apache Kafka sebagai tabel sumber atau hasil, lihat Message Queue for Apache Kafka.

• Untuk meningkatkan kinerja pekerjaan dengan menyesuaikan paralelisme node dan sumber daya, lihat Konfigurasikan penyebaran pekerjaan.