MongoDB - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Konektor MongoDB mengintegrasikan ApsaraDB for MongoDB dan MongoDB yang dikelola sendiri ke dalam Realtime Compute for Apache Flink sebagai tabel sumber, dimensi, dan sink. Konektor ini menggunakan Change Stream API untuk menangkap event insert, update, replace, dan delete secara real time.

Kemampuan

Kategori	Deskripsi
Jenis tabel	Sumber SQL, lookup (dimensi), dan sink Sumber Flink CDC Sumber DataStream
Mode berjalan	Streaming
Jenis API	DataStream API, SQL, Flink CDC
Semantik penulisan sink	Insert, update, dan delete (dengan primary key dideklarasikan)

Metrik pemantauan

Tabel sumber:

numBytesIn, numBytesInPerSecond, numRecordsIn, numRecordsInPerSecond, numRecordsInErrors, currentFetchEventTimeLag, currentEmitEventTimeLag, watermarkLag, sourceIdleTime

Tabel dimensi dan sink tidak mengekspos metrik pemantauan.

Untuk definisi metrik, lihat Metrik.

Cara kerja

Konektor MongoDB membaca data dalam dua fase:

Snapshot penuh — membaca semua dokumen yang ada dari koleksi target secara paralel.
Pembacaan inkremental — secara otomatis beralih ke mengonsumsi oplog melalui Change Stream API setelah snapshot selesai.

Proses ini menyediakan semantik tepat-sekali, memastikan tidak ada catatan duplikat atau hilang selama pemulihan kesalahan.

Konsep utama

Mode startup

Pilih mode startup berdasarkan kapan pipeline Anda perlu mulai mengonsumsi data:

Mode	Perilaku	Gunakan saat
`initial` (default)	Membaca snapshot saat pertama kali dijalankan, lalu beralih ke pembacaan inkremental	Anda memerlukan salinan lengkap data yang ada
`latest-offset`	Dimulai dari posisi oplog saat ini; tidak ada data historis	Anda hanya memerlukan perubahan mulai sekarang
`timestamp`	Membaca event oplog dari timestamp tertentu; melewati snapshot	Anda memerlukan perubahan mulai dari titik waktu tertentu (memerlukan MongoDB 4.0+)

Dukungan changelog lengkap

Secara default, MongoDB tidak menyimpan status dokumen sebelum perubahan (versi sebelum MongoDB 6.0). Tanpa informasi ini, konektor hanya dapat menghasilkan event UPSERT — catatan UPDATE_BEFORE tidak tersedia.

Sebagai solusi, planner SQL Flink menyisipkan operator ChangelogNormalize yang menyimpan status dokumen di backend status Flink. Meskipun berfungsi, pendekatan ini mengonsumsi penyimpanan status yang signifikan.

MongoDB 6.0+ mendukung pencatatan preimage dan postimage. Saat diaktifkan, MongoDB mencatat status dokumen lengkap sebelum dan sesudah setiap perubahan. Mengatur scan.full-changelog ke true memberi tahu konektor untuk menggunakan catatan ini guna menghasilkan aliran changelog lengkap — menghilangkan operator ChangelogNormalize dan beban statusnya.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Instans ApsaraDB for MongoDB (replica set atau kluster sharded), atau kluster MongoDB 3.6+ yang dikelola sendiri dengan mode replica set diaktifkan. Lihat Replication.
Jika autentikasi diaktifkan, pengguna MongoDB dengan izin berikut: splitVector, listDatabases, listCollections, collStats, find, changeStream, serta akses baca ke config.collections dan config.chunks.
Alamat IP kluster Flink ditambahkan ke daftar izin IP MongoDB.
Database dan koleksi target telah dibuat sebelum menjalankan pekerjaan.

Batasan

Sumber SQL

Pembacaan snapshot paralel memerlukan MongoDB 4.0 atau yang lebih baru. Aktifkan dengan mengatur scan.incremental.snapshot.enabled ke true.
Database admin, local, dan config serta semua koleksi sistem tidak dapat dipantau. Ini adalah batasan Change Stream MongoDB. Lihat Change Streams dalam dokumentasi MongoDB.
Saat membuat tabel sumber SQL, deklarasikan kolom _id STRING dan tetapkan sebagai primary key.

SQL sink

VVR 8.0.4 dan sebelumnya: hanya insert.
VVR 8.0.5 dan seterusnya dengan primary key dideklarasikan: insert, update, dan delete.
VVR 8.0.5 dan seterusnya tanpa primary key: hanya insert.
Pengiriman tepat-sekali tidak didukung. Opsi sink.delivery-guarantee menerima none atau at-least-once.

Lookup SQL (dimensi)

Didukung di VVR 8.0.5 dan seterusnya.
VVR 8.0.9 dan seterusnya: join lookup mendukung pembacaan field bawaan _id bertipe ObjectId.

SQL

Sintaksis

CREATE TABLE tableName(
  _id STRING,
  [columnName dataType,]*
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'localhost:27017',
  'username' = 'mongouser',
  'password' = '${secret_values.password}',
  'database' = 'testdb',
  'collection' = 'testcoll'
)

Deklarasikan kolom _id STRING dan tentukan sebagai primary key saat membuat tabel sumber CDC.

Opsi konektor

Umum

Opsi	Tipe	Wajib	Default	Deskripsi
`connector`	String	Ya	—	Identifier konektor. Tabel sumber: `mongodb-cdc` (VVR 8.0.4 dan sebelumnya) atau `mongodb` / `mongodb-cdc` (VVR 8.0.5 dan seterusnya). Tabel dimensi atau sink: `mongodb`.
`uri`	String	Tidak	—	URI koneksi MongoDB. Tentukan salah satu antara `uri` atau `hosts`. Jika Anda menentukan `uri`, abaikan `scheme`, `hosts`, `username`, `password`, dan `connection.options`. Jika keduanya diatur, `uri` memiliki prioritas lebih tinggi.
`hosts`	String	Tidak	—	Hostname server MongoDB. Pisahkan beberapa host dengan koma (`,`).
`scheme`	String	Tidak	`mongodb`	Protokol koneksi. Nilai valid: `mongodb` (default), `mongodb+srv` (DNS SRV).
`username`	String	Tidak	—	Username MongoDB. Diperlukan saat autentikasi diaktifkan.
`password`	String	Tidak	—	Password MongoDB. Diperlukan saat autentikasi diaktifkan. Gunakan variabel daripada hardcoding kredensial.
`database`	String	Tidak	—	Nama database MongoDB. Mendukung ekspresi reguler untuk tabel sumber. Jika tidak diatur, semua database dipantau. Tidak dapat memantau database `admin`, `local`, atau `config`.
`collection`	String	Tidak	—	Nama koleksi MongoDB. Mendukung ekspresi reguler untuk tabel sumber. Catatan: Jika tidak diatur, semua koleksi dipantau. Tidak dapat memantau koleksi sistem. Jika nama koleksi berisi karakter khusus ekspresi reguler, gunakan namespace berkualifikasi penuh (`database.collection`).
`connection.options`	String	Tidak	—	Opsi koneksi tambahan sebagai pasangan `key=value` yang dipisahkan oleh `\&` (misalnya, `connectTimeoutMS=12000\&socketTimeoutMS=13000`). Secara default, konektor tidak mengatur timeout koneksi socket, yang dapat menyebabkan gangguan panjang selama fluktuasi jaringan. Atur `socketTimeoutMS` ke nilai yang wajar untuk menghindari hal ini.

Sumber

Opsi	Tipe	Wajib	Default	Deskripsi
`scan.startup.mode`	String	Tidak	`initial`	Mode startup. Nilai valid: `initial`, `latest-offset`, `timestamp`. Lihat Mode startup dan Startup Properties.
`scan.startup.timestamp-millis`	Long	Kondisional	—	Timestamp awal dalam milidetik sejak epoch UNIX. Diperlukan saat `scan.startup.mode` adalah `timestamp`.
`initial.snapshotting.queue.size`	Integer	Tidak	`10240`	Ukuran maksimum antrian selama fase snapshot awal. Berlaku hanya saat `scan.startup.mode` adalah `initial`.
`batch.size`	Integer	Tidak	`1024`	Ukuran batch kursor.
`poll.max.batch.size`	Integer	Tidak	`1024`	Jumlah maksimum dokumen perubahan yang ditarik per batch selama pembacaan stream. Nilai yang lebih besar mengalokasikan buffer internal yang lebih besar.
`poll.await.time.ms`	Integer	Tidak	`1000`	Interval antara permintaan tarik data, dalam milidetik.
`heartbeat.interval.ms`	Integer	Tidak	`0`	Interval heartbeat dalam milidetik. Konektor mengirim heartbeat untuk melacak posisi oplog terbaru. Mengatur ini ke `0` menonaktifkan heartbeat. Atur ini untuk koleksi yang jarang diperbarui.
`scan.incremental.snapshot.enabled`	Boolean	Tidak	`false`	Mengaktifkan pembacaan snapshot paralel. Fitur eksperimental. Memerlukan MongoDB 4.0 atau yang lebih baru.
`scan.incremental.snapshot.chunk.size.mb`	Integer	Tidak	`64`	Ukuran chunk untuk pembacaan snapshot paralel, dalam MB. Fitur eksperimental. Berlaku hanya saat pembacaan snapshot paralel diaktifkan.
`scan.full-changelog`	Boolean	Tidak	`false`	Menghasilkan aliran changelog lengkap menggunakan catatan preimage dan postimage MongoDB. Fitur eksperimental. Memerlukan MongoDB 6.0 atau yang lebih baru dengan fitur preimage dan postimage diaktifkan.
`scan.flatten-nested-columns.enabled`	Boolean	Tidak	`false`	Mengurai field yang dipisahkan oleh `.` sebagai field dokumen BSON bersarang. Misalnya, `{"nested":{"col":true}}` dipetakan ke field bernama `nested.col`. Didukung di VVR 8.0.5 dan seterusnya.
`scan.primitive-as-string`	Boolean	Tidak	`false`	Mengurai semua tipe primitif BSON sebagai STRING. Didukung di VVR 8.0.5 dan seterusnya.
`scan.ignore-delete.enabled`	Boolean	Tidak	`false`	Mengabaikan semua event DELETE (-D), termasuk yang dihasilkan selama pengarsipan data MongoDB. Didukung di VVR 11.1 dan seterusnya.
`scan.incremental.snapshot.backfill.skip`	Boolean	Tidak	`false`	Nilai valid: true: melewati backfill selama pembacaan snapshot inkremental. false (default): tidak melewati backfill. Backfill hanya berlaku selama kueri snapshot satu chunk dan tidak mencakup seluruh fase pembacaan penuh. Saat backfill dilewati, kueri snapshot setiap chunk membaca data terbaru pada saat itu; pembaruan yang terjadi pada chunk setelah dibaca tidak digabung selama fase pembacaan penuh dan dibaca dari OpLog setelah memasuki fase inkremental. Misalnya, pembaruan pada chunk5 yang terjadi saat chunk5 sedang di-snapshot tercermin langsung dalam snapshot chunk5; jika chunk5 diperbarui setelah pembaca maju ke chunk80, pembaruan tersebut diterapkan kemudian dari OpLog selama fase inkremental. Penting Saat diaktifkan, perubahan yang terjadi selama atau setelah pemindaian chunk tetap dikirimkan dari OpLog pada fase inkremental dan mungkin diduplikasi. Hanya semantik at-least-once yang dijamin. Aktifkan ini hanya jika sink downstream mendukung penulisan idempoten berdasarkan primary key. Catatan Hanya didukung di VVR 11.1 dan seterusnya.
`initial.snapshotting.pipeline`	String	Tidak	—	Operasi pipeline agregasi MongoDB yang diterapkan selama pembacaan snapshot untuk memfilter data. Tentukan sebagai array JSON, misalnya: `[{"$match": {"closed": "false"}}]`. Berlaku hanya saat `scan.startup.mode` adalah `initial` dan konektor berjalan dalam mode Debezium. Didukung di VVR 11.1 dan seterusnya.
`initial.snapshotting.max.threads`	Integer	Tidak	—	Jumlah thread untuk replikasi snapshot. Berlaku hanya saat `scan.startup.mode` adalah `initial`. Didukung di VVR 11.1 dan seterusnya.
`initial.snapshotting.queue.size`	Integer	Tidak	`16000`	Ukuran antrian untuk snapshot awal. Berlaku hanya saat `scan.startup.mode` adalah `initial`. Didukung di VVR 11.1 dan seterusnya.
`scan.change-stream.reading.parallelism`	Integer	Tidak	`1`	Jumlah pembaca konkuren untuk Change Stream. Berlaku hanya saat `scan.incremental.snapshot.enabled` adalah `true`. Juga atur `heartbeat.interval.ms` saat menggunakan opsi ini. Didukung di VVR 11.2 dan seterusnya.
`scan.change-stream.reading.queue-size`	Integer	Tidak	`16384`	Ukuran antrian pesan untuk langganan change stream konkuren. Berlaku hanya saat `scan.change-stream.reading.parallelism` diaktifkan. Didukung di VVR 11.2 dan seterusnya.

Lookup (dimensi)

Opsi	Tipe	Wajib	Default	Deskripsi
`lookup.cache`	String	Tidak	`NONE`	Kebijakan cache. Nilai valid: `NONE` (tanpa caching), `PARTIAL` (cache hasil lookup dari database eksternal).
`lookup.max-retries`	Integer	Tidak	`3`	Upaya maksimum saat lookup gagal.
`lookup.retry.interval`	Duration	Tidak	`1s`	Interval antara upaya ulang saat lookup gagal.
`lookup.partial-cache.expire-after-access`	Duration	Tidak	—	Waktu maksimum entri cache bertahan setelah akses terakhir. Unit yang didukung: `ms`, `s`, `min`, `h`, `d`. Memerlukan `lookup.cache` = `PARTIAL`.
`lookup.partial-cache.expire-after-write`	Duration	Tidak	—	Waktu maksimum entri cache bertahan setelah ditulis. Memerlukan `lookup.cache` = `PARTIAL`.
`lookup.partial-cache.max-rows`	Long	Tidak	—	Jumlah maksimum baris dalam cache. Entri terlama dihapus saat batas tercapai. Memerlukan `lookup.cache` = `PARTIAL`.
`lookup.partial-cache.cache-missing-key`	Boolean	Tidak	`true`	Menyimpan entri null saat kunci lookup tidak memiliki catatan yang cocok. Memerlukan `lookup.cache` = `PARTIAL`.
`lookup.type-conversion.mode`	String	Tidak	`FORCE_STRING`	Strategi konversi yang digunakan saat nilai string Flink dicocokkan dengan tabel dimensi MongoDB. Nilai valid: `FORCE_STRING`: selalu mengonversi field MongoDB ke string sebelum pencocokan. Tipe field yang tidak memiliki padanan di Flink, seperti ObjectId, dapat dicocokkan, tetapi indeks pada field tersebut (jika ada) tidak dapat digunakan. `OBJECT_ID_WRAPPER`: jika nilai string dari Flink merupakan string heksa ObjectId yang valid, mencocokkannya sebagai ObjectId. Indeks pada field join dapat digunakan untuk mempercepat kueri. `NONE`: tidak melakukan penanganan khusus. Tidak ada baris yang cocok jika tipe MongoDB berbeda dari tipe kueri Flink. Catatan Parameter ini hanya didukung di VVR 11.9 dan seterusnya.

Sink

Opsi	Tipe	Wajib	Default	Deskripsi
`sink.buffer-flush.max-rows`	Integer	Tidak	`1000`	Jumlah maksimum catatan yang ditulis per batch.
`sink.buffer-flush.interval`	Duration	Tidak	`1s`	Interval flush.
`sink.delivery-guarantee`	String	Tidak	`at-least-once`	Semantik pengiriman tulis. Nilai valid: `none`, `at-least-once`. Tepat-sekali tidak didukung.
`sink.max-retries`	Integer	Tidak	`3`	Upaya maksimum saat penulisan gagal.
`sink.retry.interval`	Duration	Tidak	`1s`	Interval antara upaya ulang saat penulisan gagal.
`sink.parallelism`	Integer	Tidak	—	Paralelisme sink kustom.
`sink.delete-strategy`	String	Tidak	`CHANGELOG_STANDARD`	Strategi untuk menangani event -D dan -U. Nilai valid: `CHANGELOG_STANDARD` (menerapkan pembaruan dan penghapusan secara normal), `IGNORE_DELETE` (mengabaikan event -D; menimpa baris penuh pada -U), `PARTIAL_UPDATE` (mengabaikan event -U untuk mendukung pembaruan kolom parsial; menghapus baris pada -D), `IGNORE_ALL` (mengabaikan event -U dan -D).

Pemetaan tipe data

Sumber

Tipe BSON	Flink SQL
Int32	INT
Int64	BIGINT
Double	DOUBLE
Decimal128	DECIMAL(p, s)
Boolean	BOOLEAN
Date Timestamp	DATE
Date Timestamp	TIME
DateTime	TIMESTAMP(3), TIMESTAMP_LTZ(3)
Timestamp	TIMESTAMP(0), TIMESTAMP_LTZ(0)
String, ObjectId, UUID, Symbol, MD5, JavaScript, Regex	STRING
Binary	BYTES
Object	ROW
Array	ARRAY
DBPointer	ROW\<$ref STRING, $id STRING\>
GeoJSON Point	ROW\<type STRING, coordinates ARRAY\<DOUBLE\>\>
GeoJSON Line	ROW\<type STRING, coordinates ARRAY\<ARRAY\<DOUBLE\>\>\>

Lookup (dimensi) dan sink

Tipe BSON	Tipe Flink SQL
Int32	INT
Int64	BIGINT
Double	DOUBLE
Decimal128	DECIMAL
Boolean	BOOLEAN
DateTime	TIMESTAMP_LTZ(3)
Timestamp	TIMESTAMP_LTZ(0)
String, ObjectId	STRING
Binary	BYTES
Object	ROW
Array	ARRAY

Kolom metadata

Sumber SQL mendukung kolom metadata berikut:

Kolom metadata	Tipe	Deskripsi
`database_name`	STRING NOT NULL	Database yang berisi dokumen.
`collection_name`	STRING NOT NULL	Koleksi yang berisi dokumen.
`op_ts`	TIMESTAMP_LTZ(3) NOT NULL	Waktu dokumen berubah. Mengembalikan `0` untuk dokumen dari snapshot awal.
`row_kind`	STRING NOT NULL	Jenis event perubahan: `+I` (INSERT), `-D` (DELETE), `-U` (UPDATE_BEFORE), `+U` (UPDATE_AFTER). Didukung di VVR 11.1 dan seterusnya.

Contoh

Sumber

Contoh berikut membaca dari tabel sumber MongoDB dengan pembacaan snapshot paralel dan changelog lengkap diaktifkan, lalu menulis field yang dipilih ke sink print.

-- Tabel sumber CDC: membaca data produk dari MongoDB
-- _id harus dideklarasikan dan ditetapkan sebagai primary key
CREATE TEMPORARY TABLE mongo_source (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  db_name STRING METADATA FROM 'database_name' VIRTUAL,
  collection_name STRING METADATA VIRTUAL,
  op_ts TIMESTAMP_LTZ(3) METADATA VIRTUAL,
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection',
  'scan.incremental.snapshot.enabled' = 'true', -- Aktifkan pembacaan snapshot paralel (memerlukan MongoDB 4.0+)
  'scan.full-changelog' = 'true'                -- Aktifkan changelog lengkap (memerlukan MongoDB 6.0+ dengan preimage/postimage)
);

CREATE TEMPORARY TABLE productssink (
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price_amount DECIMAL,
  suppliers_name STRING,
  db_name STRING,
  collection_name STRING,
  op_ts TIMESTAMP_LTZ(3)
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

INSERT INTO productssink
SELECT
  name,
  weight,
  tags,
  price.amount,
  suppliers[1].name,
  db_name,
  collection_name,
  op_ts
FROM mongo_source;

Lookup (dimensi)

Contoh berikut melakukan join antara stream generator data dengan tabel dimensi MongoDB menggunakan temporal join.

CREATE TEMPORARY TABLE datagen_source (
  id STRING,
  a INT,
  b BIGINT,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE mongo_dim (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  PRIMARY KEY(_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection',
  'lookup.cache' = 'PARTIAL',                           -- Cache hasil lookup untuk performa lebih baik
  'lookup.partial-cache.expire-after-access' = '10min', -- Hapus entri cache setelah 10 menit tidak aktif
  'lookup.partial-cache.expire-after-write' = '10min',  -- Hapus entri cache 10 menit setelah ditulis
  'lookup.partial-cache.max-rows' = '100'               -- Maksimal 100 baris dalam cache
);

CREATE TEMPORARY TABLE print_sink (
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price_amount DECIMAL,
  suppliers_name STRING
) WITH (
  'connector' = 'print',
  'logger' = 'true'
);

INSERT INTO print_sink
SELECT
  T.id,
  T.a,
  T.b,
  H.name
FROM datagen_source AS T
JOIN mongo_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.id = H._id;

Sink

Contoh berikut menulis data dari generator data ke tabel sink MongoDB. Primary key dideklarasikan untuk mendukung operasi insert, update, dan delete.

CREATE TEMPORARY TABLE datagen_source (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE mongo_sink (
  `_id` STRING,
  name STRING,
  weight DECIMAL,
  tags ARRAY<STRING>,
  price ROW<amount DECIMAL, currency STRING>,
  suppliers ARRAY<ROW<name STRING, address STRING>>,
  PRIMARY KEY(_id) NOT ENFORCED -- Deklarasikan primary key untuk mengaktifkan update dan delete
) WITH (
  'connector' = 'mongodb',
  'hosts' = 'dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,dds-bp169b982fc25****.mongodb.rds.aliyuncs.com:3717,',
  'username' = 'root',
  'password' = '${secret_values.password}',
  'database' = 'flinktest',
  'collection' = 'flinkcollection'
);

INSERT INTO mongo_sink SELECT * FROM datagen_source;

Flink CDC (pratinjau publik)

Flink CDC memungkinkan Anda menyinkronkan data MongoDB ke penyimpanan downstream menggunakan pipeline berbasis skrip YAML, tanpa menulis DDL SQL. Fitur ini memerlukan VVR 11.1 dan seterusnya.

Sintaksis

source:
   type: mongodb
   name: MongoDB Source
   hosts: localhost:33076
   username: ${mongo.username}
   password: ${mongo.password}
   database: foo_db
   collection: foo_col_.*

sink:
  type: ...

Opsi konfigurasi

Opsi	Wajib	Tipe	Default	Deskripsi
`type`	Ya	STRING	—	Konektor. Atur ke `mongodb`.
`scheme`	Tidak	STRING	`mongodb`	Protokol koneksi. Nilai valid: `mongodb`, `mongodb+srv`.
`hosts`	Ya	STRING	—	Hostname server MongoDB. Pisahkan beberapa host dengan koma.
`username`	Tidak	STRING	—	Username MongoDB.
`password`	Tidak	STRING	—	Password MongoDB.
`database`	Ya	STRING	—	Nama database MongoDB yang akan ditangkap. Ekspresi reguler didukung.
`collection`	Ya	STRING	—	Nama koleksi MongoDB yang akan ditangkap. Ekspresi reguler didukung. Gunakan namespace berkualifikasi penuh `database.collection`.
`connection.options`	Tidak	STRING	—	Opsi koneksi tambahan sebagai pasangan `k=v` yang dipisahkan oleh `\&`, misalnya: `replicaSet=test\&connectTimeoutMS=300000`.
`schema.inference.strategy`	Tidak	STRING	`continuous`	Strategi inferensi skema. `continuous`: menginferensi tipe secara kontinu dan mengeluarkan event perubahan skema saat skema melebar. `static`: menginferensi skema sekali saat startup.
`scan.max.pre.fetch.records`	Tidak	INT	`50`	Jumlah maksimum catatan yang disampel per koleksi selama inferensi skema awal.
`scan.startup.mode`	Tidak	STRING	`initial`	Mode startup. Nilai valid: `initial`, `latest-offset`, `timestamp`, `snapshot`.
`scan.startup.timestamp-millis`	Tidak	LONG	—	Timestamp awal dalam milidetik. Diperlukan saat `scan.startup.mode` adalah `timestamp`.
`chunk-meta.group.size`	Tidak	INT	`1000`	Ukuran maksimum chunk metadata.
`scan.incremental.close-idle-reader.enabled`	Tidak	BOOLEAN	`false`	Menutup pembaca sumber yang idle setelah beralih ke pembacaan inkremental.
`scan.incremental.snapshot.backfill.skip`	Tidak	BOOLEAN	`false`	Nilai valid: true: melewati backfill selama pembacaan snapshot inkremental. false (default): tidak melewati backfill. Backfill hanya berlaku selama kueri snapshot satu chunk dan tidak mencakup seluruh fase pembacaan penuh. Saat backfill dilewati, kueri snapshot setiap chunk membaca data terbaru pada saat itu; pembaruan yang terjadi pada chunk setelah dibaca tidak digabung selama fase pembacaan penuh dan dibaca dari OpLog setelah memasuki fase inkremental. Misalnya, pembaruan pada chunk5 yang terjadi saat chunk5 sedang di-snapshot tercermin langsung dalam snapshot chunk5; jika chunk5 diperbarui setelah pembaca maju ke chunk80, pembaruan tersebut diterapkan kemudian dari OpLog selama fase inkremental. Penting Saat diaktifkan, perubahan yang terjadi selama atau setelah pemindaian chunk tetap dikirimkan dari OpLog pada fase inkremental dan mungkin diduplikasi. Hanya semantik at-least-once yang dijamin. Aktifkan ini hanya jika sink downstream mendukung penulisan idempoten berdasarkan primary key.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Tidak	BOOLEAN	`false`	Membaca chunk tak terbatas terlebih dahulu. Mengurangi risiko kehabisan memori untuk koleksi yang sering diperbarui.
`batch.size`	Tidak	INT	`1024`	Ukuran batch kursor.
`poll.max.batch.size`	Tidak	INT	`1024`	Jumlah maksimum entri per permintaan tarik Change Stream.
`poll.await.time.ms`	Tidak	INT	`1000`	Waktu tunggu minimum antara permintaan tarik Change Stream, dalam milidetik.
`heartbeat.interval.ms`	Tidak	INT	`0`	Interval heartbeat dalam milidetik. Atur ini untuk koleksi yang jarang diperbarui. Mengatur ke `0` menonaktifkan heartbeat.
`scan.incremental.snapshot.chunk.size.mb`	Tidak	INT	`64`	Ukuran chunk selama snapshotting, dalam MB.
`scan.incremental.snapshot.chunk.samples`	Tidak	INT	`20`	Jumlah sampel yang digunakan untuk memperkirakan ukuran koleksi selama snapshotting.
`scan.full-changelog`	Tidak	BOOLEAN	`false`	Menghasilkan event changelog lengkap menggunakan catatan preimage dan postimage. Memerlukan MongoDB 6.0 atau yang lebih baru dengan preimage dan postimage diaktifkan.
`scan.cursor.no-timeout`	Tidak	BOOLEAN	`false`	Menonaktifkan timeout kursor. Secara default, MongoDB menutup kursor idle setelah 10 menit.
`scan.ignore-delete.enabled`	Tidak	BOOLEAN	`false`	Mengabaikan event delete dari MongoDB.
`scan.flatten.nested-documents.enabled`	Tidak	BOOLEAN	`false`	Meratakan dokumen BSON bersarang. Misalnya, `{"doc": {"foo": 1, "bar": "two"}}` menjadi `doc.foo INT, doc.bar STRING`.
`scan.all.primitives.as-string.enabled`	Tidak	BOOLEAN	`false`	Menginferensi semua tipe primitif sebagai STRING. Mengurangi event perubahan skema saat tipe upstream tidak konsisten.
`metadata.list`	Tidak	STRING	—	Daftar field metadata yang dipisahkan koma untuk dikirimkan ke downstream. Nilai yang didukung: `ts_ms` (timestamp event OpLog), `op_ts` (alias untuk `ts_ms`; gunakan saat menulis metadata ke Kafka JSON).

Pemetaan tipe data

MongoDB BSON	Flink CDC	Catatan
STRING	VARCHAR	—
INT32	INT	—
INT64	BIGINT	—
DECIMAL128	DECIMAL	—
DOUBLE	DOUBLE	—
BOOLEAN	BOOLEAN	—
TIMESTAMP	TIMESTAMP	—
DATETIME	LOCALZONEDTIMESTAMP	—
BINARY	VARBINARY	—
DOCUMENT	MAP	Tipe kunci dan nilai diinferensi.
ARRAY	ARRAY	Tipe elemen diinferensi.
OBJECTID	VARCHAR	Direpresentasikan sebagai string heksa.
SYMBOL, REGULAREXPRESSION, JAVASCRIPT, JAVASCRIPTWITHSCOPE	VARCHAR	Direpresentasikan sebagai string.

Kolom metadata

Flink CDC mendukung kolom metadata berikut untuk konektor MongoDB:

Kolom metadata	Tipe	Deskripsi
`ts_ms`	BIGINT NOT NULL	Waktu dokumen berubah (timestamp OpLog). Mengembalikan `0` untuk dokumen dari snapshot awal.

Gunakan kolom metadata generik modul Transform untuk mengakses database_name, collection_name, dan row_kind.

DataStream API

Penting

Untuk menggunakan DataStream API, siapkan konektor DataStream untuk pekerjaan Anda. Lihat Penggunaan konektor DataStream.

Tambahkan dependensi Maven

Repositori Maven Central menyediakan konektor MongoDB VVR.

<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>flink-connector-mongodb</artifactId>
    <version>${vvr.version}</version>
</dependency>

Buat MongoDBSource

Gunakan MongoDBSource.builder() untuk membuat sumber:

Untuk mengaktifkan pembacaan snapshot inkremental, gunakan builder dari com.ververica.cdc.connectors.mongodb.source.
Jika tidak, gunakan builder dari com.ververica.cdc.connectors.mongodb.

MongoDBSource.builder()
  .hosts("mongo.example.com:27017")
  .username("mongouser")
  .password("mongopasswd")
  .databaseList("testdb")          // Mendukung ekspresi reguler; gunakan .* untuk mencocokkan semua database
  .collectionList("testcoll")      // Mendukung ekspresi reguler; gunakan .* untuk mencocokkan semua koleksi
  .startupOptions(StartupOptions.initial()) // StartupOptions.latest-offset(), StartupOptions.timestamp()
  .deserializer(new JsonDebeziumDeserializationSchema())
  .build();

Parameter MongoDBSource

Parameter	Deskripsi
`hosts`	Hostname server MongoDB.
`username`	Username MongoDB. Abaikan jika autentikasi tidak diaktifkan.
`password`	Password MongoDB. Abaikan jika autentikasi tidak diaktifkan.
`databaseList`	Nama database yang akan dipantau. Mendukung ekspresi reguler. Gunakan `.*` untuk mencocokkan semua database.
`collectionList`	Nama koleksi yang akan dipantau. Mendukung ekspresi reguler. Gunakan `.*` untuk mencocokkan semua koleksi.
`startupOptions`	Mode startup. Nilai valid: `StartupOptions.initial()`, `StartupOptions.latest-offset()`, `StartupOptions.timestamp()`.
`deserializer`	Deserializer untuk mengonversi objek `SourceRecord`. Nilai valid: `MongoDBConnectorDeserializationSchema` (mode upsert, menghasilkan Flink `RowData`), `MongoDBConnectorFullChangelogDeserializationSchema` (mode changelog lengkap, menghasilkan Flink `RowData`), `JsonDebeziumDeserializationSchema` (menghasilkan string JSON).

Referensi

Flink CDC (pratinjau publik) — Menyinkronkan data MongoDB dan perubahan skema ke tabel downstream (VVR 11.1 dan seterusnya).
Metrik — Memantau performa tabel sumber.