MongoDB - Realtime Compute for Apache Flink - Alibaba Cloud Dokümantasyon Merkezi

Yetenekler

Kategori	Açıklama
Tablo türleri	SQL kaynağı, arama (boyut/dimension) ve havuz (sink) Flink CDC kaynağı DataStream kaynağı
Çalıştırma modu	Akış
API türleri	DataStream API, SQL, Flink CDC
Havuz yazma semantiği	Ekleme, güncelleme ve silme (birincil anahtar bildirildiğinde)

İzleme ölçümleri

Kaynak tablolar:

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

Boyut (dimension) ve havuz (sink) tabloları izleme ölçümü sunmaz.

Ölçüm tanımları için bkz. Metrics.

Nasıl çalışır

MongoDB bağlayıcısı verileri iki aşamada okur:

Tam anlık görüntü — hedeflenen koleksiyonlardaki tüm mevcut belgeleri paralel olarak okur.
Artımlı okuma — anlık görüntü tamamlandıktan sonra Change Stream API aracılığıyla oplog'u tüketmeye otomatik olarak geçer.

Bu işlem tam olarak bir kez (exactly-once) semantiği sağlar ve arıza kurtarma sırasında yinelenen veya eksik kayıt olmamasını garanti eder.

Temel kavramlar

Başlatma modları

İşlem hattınızın veri tüketmeye ne zaman başlaması gerektiğine göre bir başlatma modu seçin:

Mod	Davranış	Ne zaman kullanılır
`initial` (varsayılan)	İlk başlatmada bir anlık görüntü okur, ardından artımlı okumaya geçer	Mevcut verilerin tam bir kopyasına ihtiyacınız olduğunda
`latest-offset`	Mevcut oplog konumundan başlar; geçmiş veriler dahil edilmez	Yalnızca şu andan sonraki değişikliklere ihtiyacınız olduğunda
`timestamp`	Belirtilen bir zaman damgasından itibaren oplog olaylarını okur; anlık görüntüyü atlar	Bilinen bir zaman noktasından başlayan değişikliklere ihtiyacınız olduğunda (MongoDB 4.0+ gerekir)

Tam değişiklik günlüğü desteği

Varsayılan olarak MongoDB, belgelerin önceki durumunu (MongoDB 6.0 öncesi sürümler) saklamaz. Bu bilgi olmadan bağlayıcı yalnızca UPSERT olayları üretebilir; UPDATE_BEFORE kayıtları eksik olur.

Bunu aşmak için Flink SQL planlayıcısı, belge durumunu Flink'in durum arka ucunda önbelleğe alan bir ChangelogNormalize işleci ekler. İşlevsel olsa da bu yaklaşım önemli miktarda durum depolama alanı tüketir.

MongoDB 6.0 ve üzeri, ön görüntü (preimage) ve son görüntü (postimage) kaydını destekler. Etkinleştirildiğinde MongoDB, her değişiklikten önce ve sonra belgenin tam durumunu kaydeder. scan.full-changelog parametresini true olarak ayarlamak, bağlayıcının tam değişiklik günlükleri üretmek için bu kayıtları kullanmasını sağlar; bu sayede ChangelogNormalize işleci ve bunun getirdiği durum yükü ortadan kalkar.

Ön koşullar

Başlamadan önce şunlara sahip olduğunuzdan emin olun:

Kopya kümesi veya parçalı küme yapılandırmasına sahip bir ApsaraDB for MongoDB örneği ya da kopya kümesi modu etkinleştirilmiş kendi kendinize yönettiğiniz MongoDB 3.6+ kümesi. Bkz. Replication.
Kimlik doğrulama etkinse aşağıdaki izinlere sahip bir MongoDB kullanıcısı: splitVector, listDatabases, listCollections, collStats, find, changeStream ve config.collections ile config.chunks üzerinde okuma erişimi.
Flink küme IP adreslerinin MongoDB IP izin listesine eklenmiş olması.
İşi çalıştırmadan önce hedef veritabanının ve koleksiyonun oluşturulmuş olması.

Sınırlamalar

SQL kaynağı

Paralel anlık görüntü okuma için MongoDB 4.0 veya üzeri gerekir. scan.incremental.snapshot.enabled parametresini true olarak ayarlayarak etkinleştirin.
admin, local ve config veritabanları ile tüm sistem koleksiyonları izlenemez. Bu, MongoDB Change Stream kısıtlamasıdır. MongoDB belgelerindeki Change Streams bölümüne bakın.
Bir SQL kaynak tablosu oluştururken _id STRING sütununu bildirin ve bunu birincil anahtar olarak ayarlayın.

SQL havuzu (sink)

VVR 8.0.4 ve önceki sürümler: yalnızca ekleme.
VVR 8.0.5 ve sonraki sürümlerde birincil anahtar bildirildiyse: ekleme, güncelleme ve silme.
VVR 8.0.5 ve sonraki sürümlerde birincil anahtar yoksa: yalnızca ekleme.
Tam olarak bir kez (exactly-once) teslimat desteklenmez. sink.delivery-guarantee seçeneği none veya at-least-once değerlerini kabul eder.

SQL arama (boyut/dimension)

VVR 8.0.5 ve sonraki sürümlerde desteklenir.
VVR 8.0.9 ve sonraki sürümler: Arama birleştirmeleri (lookup joins), ObjectId türündeki yerleşik _id alanının okunmasını destekler.

SQL

Söz dizimi

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'
)

Bir CDC kaynak tablosu oluştururken _id STRING sütununu bildirin ve bunu birincil anahtar olarak belirtin.

Bağlayıcı seçenekleri

Genel

Seçenek	Tür	Zorunlu	Varsayılan	Açıklama
`connector`	String	Evet	—	Bağlayıcı tanımlayıcısı. Kaynak tablolar: `mongodb-cdc` (VVR 8.0.4 ve önceki sürümler) veya `mongodb` / `mongodb-cdc` (VVR 8.0.5 ve sonraki sürümler). Boyut (dimension) veya havuz (sink) tabloları: `mongodb`.
`uri`	String	Hayır	—	MongoDB bağlantı URI'si. Ya `uri` ya da `hosts` belirtin. `uri` belirtirseniz `scheme`, `hosts`, `username`, `password` ve `connection.options` parametrelerini atlayın. İkisi de ayarlanırsa `uri` önceliklidir.
`hosts`	String	Hayır	—	MongoDB sunucusunun ana makine adı. Birden fazla ana makineyi virgülle (`,`) ayırın.
`scheme`	String	Hayır	`mongodb`	Bağlantı protokolü. Geçerli değerler: `mongodb` (varsayılan), `mongodb+srv` (DNS SRV).
`username`	String	Hayır	—	MongoDB kullanıcı adı. Kimlik doğrulama etkin olduğunda gereklidir.
`password`	String	Hayır	—	MongoDB parolası. Kimlik doğrulama etkin olduğunda gereklidir. Kimlik bilgilerini sabit kodlamak yerine değişkenler kullanın.
`database`	String	Hayır	—	MongoDB veritabanı adı. Kaynak tablolar için düzenli ifadeleri destekler. Ayarlanmazsa tüm veritabanları izlenir. `admin`, `local` veya `config` veritabanları izlenemez.
`collection`	String	Hayır	—	MongoDB koleksiyon adı. Kaynak tablolar için düzenli ifadeleri destekler. Notlar: Ayarlanmazsa tüm koleksiyonlar izlenir. Sistem koleksiyonları izlenemez. Koleksiyon adı düzenli ifade özel karakterleri içeriyorsa tam nitelikli ad alanını (`database.collection`) kullanın.
`connection.options`	String	Hayır	—	`\&` ile ayrılmış `key=value` çiftleri biçiminde ek bağlantı seçenekleri (örneğin, `connectTimeoutMS=12000\&socketTimeoutMS=13000`). Varsayılan olarak bağlayıcı soket bağlantı zaman aşımı ayarlamaz; bu durum ağ dalgalanmaları sırasında uzun kesintilere neden olabilir. Bunu önlemek için `socketTimeoutMS` parametresini makul bir değere ayarlayın.

Kaynak

Seçenek	Tür	Zorunlu	Varsayılan	Açıklama
`scan.startup.mode`	String	Hayır	`initial`	Başlatma modu. Geçerli değerler: `initial`, `latest-offset`, `timestamp`. Bkz. Startup modes ve Startup Properties.
`scan.startup.timestamp-millis`	Long	Koşullu	—	UNIX epoch'tan itibaren milisaniye cinsinden başlangıç zaman damgası. `scan.startup.mode` parametresi `timestamp` olduğunda gereklidir.
`initial.snapshotting.queue.size`	Integer	Hayır	`10240`	İlk anlık görüntü aşamasındaki maksimum kuyruk boyutu. Yalnızca `scan.startup.mode` parametresi `initial` olduğunda etkili olur.
`batch.size`	Integer	Hayır	`1024`	İmleç toplu iş boyutu.
`poll.max.batch.size`	Integer	Hayır	`1024`	Akış okuma sırasında toplu iş başına çekilen maksimum değişiklik belgesi sayısı. Daha büyük değerler daha büyük bir dahili arabellek ayırır.
`poll.await.time.ms`	Integer	Hayır	`1000`	Veri çekme istekleri arasındaki aralık (milisaniye).
`heartbeat.interval.ms`	Integer	Hayır	`0`	Milisaniye cinsinden kalp atışı aralığı. Bağlayıcı, en son oplog konumunu izlemek için kalp atışı gönderir. Bu parametreyi `0` olarak ayarlamak kalp atışlarını devre dışı bırakır. Nadiren güncellenen koleksiyonlar için ayarlayın.
`scan.incremental.snapshot.enabled`	Boolean	Hayır	`false`	Paralel anlık görüntü okumayı etkinleştirir. Deneysel özelliktir. MongoDB 4.0 veya üzeri gerekir.
`scan.incremental.snapshot.chunk.size.mb`	Integer	Hayır	`64`	Paralel anlık görüntü okuma için parça boyutu (MB). Deneysel özelliktir. Yalnızca paralel anlık görüntü okuma etkinleştirildiğinde etkili olur.
`scan.full-changelog`	Boolean	Hayır	`false`	MongoDB ön görüntü (preimage) ve son görüntü (postimage) kayıtlarını kullanarak tam değişiklik günlüğü akışı oluşturur. Deneysel özelliktir. Ön görüntü ve son görüntü özellikleri etkinleştirilmiş MongoDB 6.0 veya üzeri gerekir.
`scan.flatten-nested-columns.enabled`	Boolean	Hayır	`false`	`.` ile ayrılan alanları iç içe geçmiş BSON belge alanları olarak ayrıştırır. Örneğin, `{"nested":{"col":true}}` ifadesi `nested.col` adlı bir alana eşlenir. VVR 8.0.5 ve sonraki sürümlerde desteklenir.
`scan.primitive-as-string`	Boolean	Hayır	`false`	Tüm ilkel BSON türlerini STRING olarak ayrıştırır. VVR 8.0.5 ve sonraki sürümlerde desteklenir.
`scan.ignore-delete.enabled`	Boolean	Hayır	`false`	MongoDB veri arşivleme sırasında oluşanlar dahil olmak üzere tüm DELETE (-D) olaylarını yoksayar. VVR 11.1 ve sonraki sürümlerde desteklenir.
`scan.incremental.snapshot.backfill.skip`	Boolean	Hayır	`false`	Geçerli değerler: true: Artımlı anlık görüntü okuma sırasında geri doldurmayı (backfill) atlar. false (varsayılan): Geri doldurmayı atlamaz. Geri doldurma yalnızca tek bir parçanın anlık görüntü sorgusu sırasında uygulanır ve tüm tam okuma fazını kapsamaz. Geri doldurma atlandığında, her parçanın anlık görüntü sorgusu o andaki en son verileri okur; bir parça okunduktan sonra o parçada gerçekleşen güncellemeler tam okuma fazı sırasında birleştirilmez ve artımlı faza geçildikten sonra OpLog'dan okunur. Örneğin, chunk5 anlık görüntülenirken chunk5 üzerinde yapılan bir güncelleme doğrudan chunk5'in anlık görüntüsünde yansıtılır; ancak okuyucu chunk80'e ilerledikten sonra chunk5 güncellenirse bu güncelleme artımlı faz sırasında OpLog'dan daha sonra uygulanır. Önemli Etkinleştirildiğinde, bir parçanın taraması sırasında veya sonrasında gerçekleşen değişiklikler artımlı fazda OpLog'dan yine de iletilir ve yinelenebilir. Yalnızca en az bir kez (at-least-once) semantiği garanti edilir. Bu seçeneği yalnızca aşağı akış havuzu birincil anahtara göre idempotent yazmaları destekliyorsa etkinleştirin. Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
`initial.snapshotting.pipeline`	String	Hayır	—	Anlık görüntü okuma sırasında verileri filtrelemek için uygulanan MongoDB toplama işlem hattı (aggregation pipeline) işlemleri. JSON dizisi olarak belirtin, örneğin: `[{"$match": {"closed": "false"}}]`. Yalnızca `scan.startup.mode` parametresi `initial` olduğunda ve bağlayıcı Debezium modunda çalışırken etkili olur. VVR 11.1 ve sonraki sürümlerde desteklenir.
`initial.snapshotting.max.threads`	Integer	Hayır	—	Anlık görüntü çoğaltma için kullanılan iş parçacığı sayısı. Yalnızca `scan.startup.mode` parametresi `initial` olduğunda etkili olur. VVR 11.1 ve sonraki sürümlerde desteklenir.
`initial.snapshotting.queue.size`	Integer	Hayır	`16000`	İlk anlık görüntü için kuyruk boyutu. Yalnızca `scan.startup.mode` parametresi `initial` olduğunda etkili olur. VVR 11.1 ve sonraki sürümlerde desteklenir.
`scan.change-stream.reading.parallelism`	Integer	Hayır	`1`	Change Stream için eşzamanlı okuyucu sayısı. Yalnızca `scan.incremental.snapshot.enabled` parametresi `true` olduğunda etkili olur. Bu seçeneği kullanırken `heartbeat.interval.ms` parametresini de ayarlayın. VVR 11.2 ve sonraki sürümlerde desteklenir.
`scan.change-stream.reading.queue-size`	Integer	Hayır	`16384`	Eşzamanlı değişiklik akışı abonelikleri için ileti kuyruğu boyutu. Yalnızca `scan.change-stream.reading.parallelism` etkinleştirildiğinde etkili olur. VVR 11.2 ve sonraki sürümlerde desteklenir.

Arama (boyut/dimension)

Seçenek	Tür	Zorunlu	Varsayılan	Açıklama
`lookup.cache`	String	Hayır	`NONE`	Önbellek ilkesi. Geçerli değerler: `NONE` (önbellekleme yok), `PARTIAL` (harici veritabanından gelen arama sonuçlarını önbelleğe alır).
`lookup.max-retries`	Integer	Hayır	`3`	Arama başarısız olduğunda maksimum yeniden deneme sayısı.
`lookup.retry.interval`	Duration	Hayır	`1s`	Arama başarısız olduğunda yeniden denemeler arasındaki aralık.
`lookup.partial-cache.expire-after-access`	Duration	Hayır	—	Son erişimden sonra önbelleğe alınmış bir girdinin yaşayabileceği maksimum süre. Desteklenen birimler: `ms`, `s`, `min`, `h`, `d`. `lookup.cache` = `PARTIAL` olmasını gerektirir.
`lookup.partial-cache.expire-after-write`	Duration	Hayır	—	Yazıldıktan sonra önbelleğe alınmış bir girdinin yaşayabileceği maksimum süre. `lookup.cache` = `PARTIAL` olmasını gerektirir.
`lookup.partial-cache.max-rows`	Long	Hayır	—	Önbellekteki maksimum satır sayısı. Limit aşıldığında en eski girdiler temizlenir. `lookup.cache` = `PARTIAL` olmasını gerektirir.
`lookup.partial-cache.cache-missing-key`	Boolean	Hayır	`true`	Bir arama anahtarının eşleşen bir kaydı olmadığında null girdisini önbelleğe alır. `lookup.cache` = `PARTIAL` olmasını gerektirir.
`lookup.type-conversion.mode`	String	Hayır	`FORCE_STRING`	Flink dize değerleri MongoDB boyut tablosuyla eşleştirilirken kullanılan dönüştürme stratejisi. Geçerli değerler: `FORCE_STRING`: Eşleştirmeden önce MongoDB alanlarını her zaman dizeye dönüştürür. ObjectId gibi Flink karşılığı olmayan alan türleri eşleştirilebilir ancak alan üzerindeki dizin (varsa) kullanılamaz. `OBJECT_ID_WRAPPER`: Flink'ten gelen bir dize değeri geçerli bir ObjectId onaltılık dizesiyse bunu ObjectId olarak eşleştirir. Birleştirme alanındaki dizin sorguyu hızlandırmak için kullanılabilir. `NONE`: Özel bir işleme yapmaz. MongoDB türü Flink sorgu türünden farklıysa hiçbir satır eşleşmez. Not Bu parametre yalnızca VVR 11.9 ve sonraki sürümlerde desteklenir.

Havuz (sink)

Seçenek	Tür	Zorunlu	Varsayılan	Açıklama
`sink.buffer-flush.max-rows`	Integer	Hayır	`1000`	Toplu iş başına yazılan maksimum kayıt sayısı.
`sink.buffer-flush.interval`	Duration	Hayır	`1s`	Temizleme aralığı.
`sink.delivery-guarantee`	String	Hayır	`at-least-once`	Yazma teslim semantiği. Geçerli değerler: `none`, `at-least-once`. Exactly-once desteklenmez.
`sink.max-retries`	Integer	Hayır	`3`	Yazma başarısız olduğunda maksimum yeniden deneme sayısı.
`sink.retry.interval`	Duration	Hayır	`1s`	Yazma başarısız olduğunda yeniden denemeler arasındaki aralık.
`sink.parallelism`	Integer	Hayır	—	Özel havuz paralelliği.
`sink.delete-strategy`	String	Hayır	`CHANGELOG_STANDARD`	-D ve -U olaylarını işleme stratejisi. Geçerli değerler: `CHANGELOG_STANDARD` (güncellemeleri ve silmeleri normal şekilde uygular), `IGNORE_DELETE` (-D olaylarını yoksayar; -U olaylarında tam satırların üzerine yazar), `PARTIAL_UPDATE` (kısmi sütun güncellemelerini desteklemek için -U olaylarını yoksayar; -D olaylarında satırları siler), `IGNORE_ALL` (hem -U hem de -D olaylarını yoksayar).

Veri türü eşlemeleri

Kaynak

BSON türü	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\>\>\>

Arama (boyut/dimension) ve havuz (sink)

BSON türü	Flink SQL türü
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

Meta veri sütunları

SQL kaynağı aşağıdaki meta veri sütunlarını destekler:

Meta veri sütunu	Tür	Açıklama
`database_name`	STRING NOT NULL	Belgeyi içeren veritabanı.
`collection_name`	STRING NOT NULL	Belgeyi içeren koleksiyon.
`op_ts`	TIMESTAMP_LTZ(3) NOT NULL	Belgenin değiştiği zaman. İlk anlık görüntüdeki belgeler için `0` döndürür.
`row_kind`	STRING NOT NULL	Değişiklik olayı türü: `+I` (INSERT), `-D` (DELETE), `-U` (UPDATE_BEFORE), `+U` (UPDATE_AFTER). VVR 11.1 ve sonraki sürümlerde desteklenir.

Örnekler

Kaynak

Aşağıdaki örnekte, paralel anlık görüntü okuma ve tam değişiklik günlüğü etkinleştirilmiş bir MongoDB kaynak tablosundan veri okunur ve seçilen alanlar bir print havuzuna yazılır.

-- CDC source table: reads product data from MongoDB
-- _id must be declared and set as the 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', -- Enable parallel snapshot reading (requires MongoDB 4.0+)
  'scan.full-changelog' = 'true'                -- Enable full changelog (requires MongoDB 6.0+ with 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;

Arama (boyut/dimension)

Aşağıdaki örnekte, bir veri oluşturucu akışı zamansal birleştirme (temporal join) kullanılarak bir MongoDB boyut tablosuyla birleştirilir.

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 lookup results for better performance
  'lookup.partial-cache.expire-after-access' = '10min', -- Evict cached entries after 10 minutes of inactivity
  'lookup.partial-cache.expire-after-write' = '10min',  -- Evict cached entries 10 minutes after write
  'lookup.partial-cache.max-rows' = '100'               -- Maximum 100 rows in the 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;

Havuz (sink)

Aşağıdaki örnekte, bir veri oluşturucudan gelen veriler bir MongoDB havuz tablosuna yazılır. Ekleme, güncelleme ve silme işlemlerini desteklemek için birincil anahtar bildirilmiştir.

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 -- Declare primary key to enable update and 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 (genel ön izleme)

Flink CDC, SQL DDL yazmadan YAML betiği tabanlı bir işlem hattı kullanarak MongoDB verilerini aşağı akıştaki bir depoya senkronize etmenizi sağlar. Bu özellik VVR 11.1 ve sonraki sürümleri gerektirir.

Söz dizimi

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

sink:
  type: ...

Yapılandırma seçenekleri

Seçenek	Zorunlu	Tür	Varsayılan	Açıklama
`type`	Evet	STRING	—	Bağlayıcı. `mongodb` olarak ayarlayın.
`scheme`	Hayır	STRING	`mongodb`	Bağlantı protokolü. Geçerli değerler: `mongodb`, `mongodb+srv`.
`hosts`	Evet	STRING	—	MongoDB sunucu ana makine adı(ları). Birden fazla ana makineyi virgülle ayırın.
`username`	Hayır	STRING	—	MongoDB kullanıcı adı.
`password`	Hayır	STRING	—	MongoDB parolası.
`database`	Evet	STRING	—	Yakalanacak MongoDB veritabanı adı. Düzenli ifadeler desteklenir.
`collection`	Evet	STRING	—	Yakalanacak MongoDB koleksiyon adı. Düzenli ifadeler desteklenir. Tam nitelikli `database.collection` ad alanını kullanın.
`connection.options`	Hayır	STRING	—	`\&` ile ayrılmış `k=v` çiftleri biçiminde ek bağlantı seçenekleri, örneğin: `replicaSet=test\&connectTimeoutMS=300000`.
`schema.inference.strategy`	Hayır	STRING	`continuous`	Şema çıkarım stratejisi. `continuous`: Türleri sürekli olarak çıkarır ve şema genişlediğinde şema değişikliği olayları yayımlar. `static`: Şemayı başlangıçta bir kez çıkarır.
`scan.max.pre.fetch.records`	Hayır	INT	`50`	İlk şema çıkarımı sırasında koleksiyon başına örneklenen maksimum kayıt sayısı.
`scan.startup.mode`	Hayır	STRING	`initial`	Başlatma modu. Geçerli değerler: `initial`, `latest-offset`, `timestamp`, `snapshot`.
`scan.startup.timestamp-millis`	Hayır	LONG	—	Milisaniye cinsinden başlangıç zaman damgası. `scan.startup.mode` parametresi `timestamp` olduğunda gereklidir.
`chunk-meta.group.size`	Hayır	INT	`1000`	Maksimum meta veri parçası boyutu.
`scan.incremental.close-idle-reader.enabled`	Hayır	BOOLEAN	`false`	Artımlı okumaya geçtikten sonra boşta kalan kaynak okuyucuları kapatır.
`scan.incremental.snapshot.backfill.skip`	Hayır	BOOLEAN	`false`	Geçerli değerler: true: Artımlı anlık görüntü okuma sırasında geri doldurmayı (backfill) atlar. false (varsayılan): Geri doldurmayı atlamaz. Geri doldurma yalnızca tek bir parçanın anlık görüntü sorgusu sırasında uygulanır ve tüm tam okuma fazını kapsamaz. Geri doldurma atlandığında, her parçanın anlık görüntü sorgusu o andaki en son verileri okur; bir parça okunduktan sonra o parçada gerçekleşen güncellemeler tam okuma fazı sırasında birleştirilmez ve artımlı faza geçildikten sonra OpLog'dan okunur. Örneğin, chunk5 anlık görüntülenirken chunk5 üzerinde yapılan bir güncelleme doğrudan chunk5'in anlık görüntüsünde yansıtılır; ancak okuyucu chunk80'e ilerledikten sonra chunk5 güncellenirse bu güncelleme artımlı faz sırasında OpLog'dan daha sonra uygulanır. Önemli Etkinleştirildiğinde, bir parçanın taraması sırasında veya sonrasında gerçekleşen değişiklikler artımlı fazda OpLog'dan yine de iletilir ve yinelenebilir. Yalnızca en az bir kez (at-least-once) semantiği garanti edilir. Bu seçeneği yalnızca aşağı akış havuzu birincil anahtara göre idempotent yazmaları destekliyorsa etkinleştirin.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Hayır	BOOLEAN	`false`	Sınırsız parçaları önce okur. Sık güncellenen koleksiyonlar için bellek yetersizliği riskini azaltır.
`batch.size`	Hayır	INT	`1024`	İmleç toplu iş boyutu.
`poll.max.batch.size`	Hayır	INT	`1024`	Change Stream çekme isteği başına maksimum girdi sayısı.
`poll.await.time.ms`	Hayır	INT	`1000`	Change Stream çekme istekleri arasındaki minimum bekleme süresi (milisaniye).
`heartbeat.interval.ms`	Hayır	INT	`0`	Milisaniye cinsinden kalp atışı aralığı. Nadiren güncellenen koleksiyonlar için ayarlayın. `0` olarak ayarlamak kalp atışlarını devre dışı bırakır.
`scan.incremental.snapshot.chunk.size.mb`	Hayır	INT	`64`	Anlık görüntü alma sırasında parça boyutu (MB).
`scan.incremental.snapshot.chunk.samples`	Hayır	INT	`20`	Anlık görüntü alma sırasında koleksiyon boyutunu tahmin etmek için kullanılan örnek sayısı.
`scan.full-changelog`	Hayır	BOOLEAN	`false`	Ön görüntü (preimage) ve son görüntü (postimage) kayıtlarını kullanarak tam değişiklik günlüğü olayları oluşturur. Ön görüntü ve son görüntü özellikleri etkinleştirilmiş MongoDB 6.0 veya üzeri gerekir.
`scan.cursor.no-timeout`	Hayır	BOOLEAN	`false`	İmleç zaman aşımını devre dışı bırakır. Varsayılan olarak MongoDB, 10 dakika boyunca boşta kalan imleçleri kapatır.
`scan.ignore-delete.enabled`	Hayır	BOOLEAN	`false`	MongoDB'den gelen silme olaylarını yoksayar.
`scan.flatten.nested-documents.enabled`	Hayır	BOOLEAN	`false`	İç içe geçmiş BSON belgelerini düzleştirir. Örneğin, `{"doc": {"foo": 1, "bar": "two"}}` ifadesi `doc.foo INT, doc.bar STRING` haline gelir.
`scan.all.primitives.as-string.enabled`	Hayır	BOOLEAN	`false`	Tüm ilkel türleri STRING olarak çıkarır. Yukarı akış türleri tutarsız olduğunda şema değişikliği olaylarını azaltır.
`metadata.list`	Hayır	STRING	—	Aşağı akışa iletilecek meta veri alanlarının virgülle ayrılmış listesi. Desteklenen değerler: `ts_ms` (OpLog olay zaman damgası), `op_ts` (`ts_ms` için takma ad; meta verileri Kafka JSON'a yazarken kullanın).

Veri türü eşlemeleri

MongoDB BSON	Flink CDC	Notlar
STRING	VARCHAR	—
INT32	INT	—
INT64	BIGINT	—
DECIMAL128	DECIMAL	—
DOUBLE	DOUBLE	—
BOOLEAN	BOOLEAN	—
TIMESTAMP	TIMESTAMP	—
DATETIME	LOCALZONEDTIMESTAMP	—
BINARY	VARBINARY	—
DOCUMENT	MAP	Anahtar ve değer türleri çıkarılır.
ARRAY	ARRAY	Öğe türleri çıkarılır.
OBJECTID	VARCHAR	Onaltılık dize olarak temsil edilir.
SYMBOL, REGULAREXPRESSION, JAVASCRIPT, JAVASCRIPTWITHSCOPE	VARCHAR	Dize olarak temsil edilir.

Meta veri sütunları

Flink CDC, MongoDB bağlayıcısı için aşağıdaki meta veri sütununu destekler:

Meta veri sütunu	Tür	Açıklama
`ts_ms`	BIGINT NOT NULL	Belgenin değiştiği zaman (OpLog zaman damgası). İlk anlık görüntüdeki belgeler için `0` döndürür.

database_name, collection_name ve row_kind alanlarına erişmek için Transform modülünün genel meta veri sütunlarını kullanın.

DataStream API

Önemli

DataStream API'yi kullanmak için işiniz için DataStream bağlayıcısını kurun. Bkz. DataStream connector usage.

Maven bağımlılığını ekleyin

Maven Central Repository, VVR MongoDB connectors barındırır.

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

MongoDBSource oluşturun

Bir kaynak oluşturmak için MongoDBSource.builder() kullanın:

Artımlı anlık görüntü okumayı etkinleştirmek için com.ververica.cdc.connectors.mongodb.source paketindeki oluşturucuyu kullanın.
Aksi takdirde com.ververica.cdc.connectors.mongodb paketindeki oluşturucuyu kullanın.

MongoDBSource.builder()
  .hosts("mongo.example.com:27017")
  .username("mongouser")
  .password("mongopasswd")
  .databaseList("testdb")          // Supports regular expressions; use .* to match all databases
  .collectionList("testcoll")      // Supports regular expressions; use .* to match all collections
  .startupOptions(StartupOptions.initial()) // StartupOptions.latest-offset(), StartupOptions.timestamp()
  .deserializer(new JsonDebeziumDeserializationSchema())
  .build();

MongoDBSource parametreleri

Parametre	Açıklama
`hosts`	MongoDB sunucusunun ana makine adı.
`username`	MongoDB kullanıcı adı. Kimlik doğrulama etkin değilse atlayın.
`password`	MongoDB parolası. Kimlik doğrulama etkin değilse atlayın.
`databaseList`	İzlenecek veritabanı adı. Düzenli ifadeleri destekler. Tüm veritabanlarını eşleştirmek için `.*` kullanın.
`collectionList`	İzlenecek koleksiyon adı. Düzenli ifadeleri destekler. Tüm koleksiyonları eşleştirmek için `.*` kullanın.
`startupOptions`	Başlatma modu. Geçerli değerler: `StartupOptions.initial()`, `StartupOptions.latest-offset()`, `StartupOptions.timestamp()`.
`deserializer`	`SourceRecord` nesnelerini dönüştürmek için seri durumdan çıkartıcı. Geçerli değerler: `MongoDBConnectorDeserializationSchema` (upsert modu, Flink `RowData` üretir), `MongoDBConnectorFullChangelogDeserializationSchema` (tam değişiklik günlüğü modu, Flink `RowData` üretir), `JsonDebeziumDeserializationSchema` (JSON dizeleri üretir).

Referanslar

Flink CDC (public preview) — MongoDB verilerini ve şema değişikliklerini aşağı akış tablolarına senkronize edin (VVR 11.1 ve sonraki sürümler).
Metrics — Kaynak tablo performansını izleyin.