Realtime Compute for Apache Flink:Manage MongoDB catalogs

Informasi latar belakang

Topik ini mencakup operasi berikut:

• Buat katalog MongoDB

• Tampilkan katalog MongoDB

• Gunakan katalog MongoDB

• Hapus katalog MongoDB

Sebelum memulai

Konfirmasikan hal-hal berikut sebelum bekerja dengan katalog MongoDB:

• Ruang kerja Realtime Compute for Apache Flink Anda menggunakan VVR 8.0.5 atau versi yang lebih baru. Diperlukan VVR 8.0.6 atau versi yang lebih baru jika Anda berencana menggunakan pernyataan CREATE TABLE AS atau CREATE DATABASE AS.

Batasan

• Hanya Realtime Compute for Apache Flink dengan VVR 8.0.5 atau versi yang lebih baru yang mendukung katalog MongoDB.

• Anda tidak dapat memodifikasi katalog MongoDB yang sudah ada menggunakan pernyataan DDL. Untuk memperbarui pengaturan koneksi, hapus katalog tersebut lalu buat ulang.

• Katalog MongoDB hanya mendukung akses baca. Anda tidak dapat membuat, memodifikasi, atau menghapus database dan tabel melalui katalog MongoDB.

Buat katalog MongoDB

1. Pada editor kode di tab Scripts pada halaman SQL Editor, masukkan pernyataan untuk membuat katalog MongoDB.

   CREATE CATALOG <yourcatalogname> WITH(
    'type'='mongodb',
    'default-database'='<dbName>',
    'hosts'='<hosts>',
    'scheme'='<scheme>',
    'username'='<username>',
    'password'='<password>',
    'connection.options'='<connectionOptions>',
    'max.fetch.records'='100',
    'scan.flatten-nested-columns.enabled'='<flattenNestedColumns>',
    'scan.primitive-as-string'='<primitiveAsString>'
   );

   Tabel berikut menjelaskan parameter-parameter tersebut.

   Parameter	Tipe data	Deskripsi	Wajib	Nilai default	Keterangan
   yourcatalogname	STRING	Nama katalog MongoDB.	Ya	N/A	Masukkan nama kustom.
   type	STRING	Jenis katalog.	Ya	N/A	Tetapkan nilainya ke mongodb.
   hosts	STRING	Hostname instans MongoDB.	Ya	N/A	Pisahkan beberapa hostname dengan koma (,).
   default-database	STRING	Nama database MongoDB default.	Ya	N/A	N/A.
   scheme	STRING	Protokol koneksi untuk database MongoDB.	Tidak	mongodb	Nilai yang valid: mongodb (menggunakan protokol MongoDB standar) mongodb+srv (menggunakan protokol rekaman DNS SRV)
   username	STRING	Username untuk menghubungkan ke database MongoDB.	Tidak	N/A	Diperlukan jika autentikasi diaktifkan pada database MongoDB.
   password	STRING	Password untuk menghubungkan ke database MongoDB.	Tidak	N/A	Diperlukan jika autentikasi diaktifkan pada database MongoDB. Untuk mencegah paparan kredensial, gunakan fitur manajemen kunci alih-alih menentukan password dalam bentuk teks biasa. Lihat Manage variables and keys.
   connection.options	STRING	Parameter koneksi tambahan untuk database MongoDB.	Tidak	N/A	Tentukan sebagai pasangan kunci-nilai dalam format key=value, dipisahkan oleh tanda ampersand (&). Contoh: connectTimeoutMS=12000&socketTimeoutMS=13000.
   max.fetch.records	INT	Jumlah maksimum dokumen yang dicoba dibaca oleh katalog MongoDB saat menginferensi skema koleksi.	Tidak	100	N/A.
   scan.flatten-nested-columns.enabled	BOOLEAN	Menentukan apakah kolom bersarang diperluas secara rekursif saat mengurai dokumen berformat BSON.	Tidak	false	Nilai yang valid: true: Kolom bersarang diperluas; jalur kolom menjadi nama kolom, misalnya {"nested": {"col": true}} menjadi nested.col. false: Dokumen bersarang diurai sebagai STRING. Penting Parameter ini hanya berlaku untuk tabel katalog MongoDB yang digunakan sebagai tabel sumber dalam penerapan Flink SQL.
   scan.primitive-as-string	BOOLEAN	Menentukan apakah semua tipe data primitif diinferensi sebagai STRING saat mengurai dokumen berformat BSON.	Tidak	false	Nilai yang valid: true: Semua tipe primitif diinferensi sebagai STRING. false: Tipe diinferensi berdasarkan pemetaan tipe BSON ke Flink. Untuk pemetaan tipe, lihat Deskripsi inferensi skema.

2. Pilih kode untuk membuat katalog dan klik Run di sisi kiri editor kode.

   

3. Di panel Catalogs di sisi kiri halaman Catalog List, verifikasi bahwa katalog baru muncul.

Tampilkan katalog MongoDB

1. Pada editor kode di tab Scripts pada halaman SQL Editor, masukkan pernyataan berikut:

   DESCRIBE `${catalog_name}`.`${db_name}`.`${collection_name}`;

   Parameter	Deskripsi
   ${catalog_name}	Nama katalog MongoDB.
   ${db_name}	Nama database ApsaraDB for MongoDB.
   ${collection_name}	Nama koleksi ApsaraDB for MongoDB.

2. Pilih kode tersebut dan klik Run di sisi kiri editor kode.

   Setelah pernyataan dijalankan, detail tabel akan muncul di panel hasil.

   

Gunakan katalog MongoDB

INSERT INTO ${other_sink_table}
SELECT...
FROM `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
/*+OPTIONS('scan.incremental.snapshot.enabled'='true')*/;

CREATE TABLE IF NOT EXISTS `${target_table_name}`
WITH(...)
AS TABLE `${mongodb_catalog}`.`${db_name}`.`${collection_name}`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table0`
AS TABLE `mongodb-catalog`.`database`.`collection0`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table1`
AS TABLE `mongodb-catalog`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `some_catalog`.`some_database`.`some_table2`
AS TABLE `mongodb-catalog`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

END;

CREATE DATABASE IF NOT EXISTS `some_catalog`.`some_database`
AS DATABASE `mongodb-catalog`.`database`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true', 'scan.full-changelog'='true') */;

INSERT INTO ${other_sink_table}
SELECT ...
FROM ${other_source_table} AS e
JOIN `${mysql_catalog}`.`${db_name}`.`${table_name}` FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

INSERT INTO `${mysql_catalog}`.`${db_name}`.`${table_name}`
SELECT ...
FROM ${other_source_table}

Hapus katalog MongoDB

1. Pada editor kode di tab Scripts pada halaman SQL Editor, masukkan pernyataan berikut:

   DROP CATALOG ${catalog_name};

   ${catalog_name} menentukan nama katalog MongoDB yang ingin Anda hapus.

2. Klik kanan pernyataan tersebut dan pilih Run dari menu pintasan untuk menghapus katalog.

3. Di panel Catalogs di sisi kiri halaman Catalog List, verifikasi bahwa katalog tersebut tidak lagi muncul.

Deskripsi inferensi skema

Saat katalog MongoDB menginferensi skema tabel, katalog tersebut secara otomatis menambahkan parameter default dan informasi primary key. Saat katalog mengurai dokumen berformat BSON untuk mendapatkan skema koleksi, katalog tersebut mencoba membaca hingga jumlah catatan data yang ditentukan oleh parameter max.fetch.records. Katalog tersebut mengurai skema setiap catatan dan menggabungkan semua skema yang diurai menjadi satu skema koleksi.

Skema koleksi terdiri dari bagian-bagian berikut:

• Kolom fisik: Katalog MongoDB menginferensi kolom fisik dari dokumen berformat BSON.

• Primary key default: Untuk setiap tabel dalam katalog MongoDB, kolom _id berfungsi sebagai primary key untuk mencegah data duplikat.

Referensi

• Untuk informasi selengkapnya tentang cara menggunakan MongoDB connector, lihat MongoDB connector.

• Jika katalog bawaan Realtime Compute for Apache Flink tidak memenuhi kebutuhan Anda, Anda dapat menggunakan katalog kustom. Untuk informasi selengkapnya, lihat Manage custom catalogs.