Realtime Compute for Apache Flink:MySQL SQL bağlayıcısı

MySQL bağlayıcısı aşağıdakileri destekler.

• Hem tam hem de artımlı verileri okumayı destekleyen ve iki ayrı işlemi sürdürme ihtiyacını ortadan kaldıran birleşik toplu ve akış işleme.

• Yatay performans ölçeklendirmesi için tam verilerin eşzamanlı okunması.

• Tam veri okumadan artımlı veri okumaya sorunsuz geçiş ve hesaplama kaynaklarını tasarruf etmek için otomatik ölçek küçültme.

• Geliştirilmiş kararlılık için tam veri okuma aşaması sırasında kaldığı yerden devam edebilen veri aktarımı.

• Çevrimiçi hizmetleri etkilemeyen, kilit gerektirmeyen tam veri okuma.

• ApsaraDB RDS for MySQL yedekleme günlüklerinin okunmasını destekleme.

• Daha düşük okuma gecikme süresi için ikili günlük dosyalarının paralel ayrıştırılması.

SQL

MySQL bağlayıcısını SQL işlerinde kaynak tablo, boyut tablosu veya hedef tablo olarak kullanabilirsiniz.

Sözdizimi

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

Parametrelerle

• Genel

  Parameter	Description	Required	Data type	Default value	Notes
  connector	Tablo türü.	Yes	STRING	None	Kaynak tablo olarak kullanıldığında bu parametreyi mysql-cdc veya mysql olarak ayarlayabilirsiniz; ikisi eşdeğerdir. Boyut tablosu veya havuz (sink) tablosu olarak kullanıldığında değer mutlaka mysql olmalıdır.
  hostname	MySQL veritabanının IP adresi veya ana bilgisayar adı.	Yes	STRING	None	Bir sanal özel cloud (VPC) adresi belirtmenizi öneririz. Not MySQL veritabanı ile Realtime Compute for Apache Flink aynı VPC içinde değilse, veritabanına erişmek için VPC'ler arası bir ağ bağlantısı kurmanız veya genel bir uç nokta kullanmanız gerekir. Daha fazla bilgi için bkz. Manage and operate workspaces ve How can a fully managed Flink cluster access the Internet?.
  username	MySQL veritabanı hizmetinin kullanıcı adı.	Yes	STRING	None	Yok.
  password	MySQL veritabanı hizmetinin parolası.	Yes	STRING	None	Yok.
  database-name	MySQL veritabanının adı.	Yes	STRING	None	Bir veritabanı kaynak tablo olarak kullanıldığında, birden çok veritabanından veri okumak için veritabanı adında düzenli ifade kullanabilirsiniz. Düzenli ifadeler kullanırken dizenin başını ve sonunu eşleştirmek üzere ^ ve $ simgelerini kullanmayın. Daha fazla bilgi için table-name parametresine ilişkin notlara bakın.
  table-name	MySQL tablosunun adı.	Yes	STRING	None	Birden çok tablodan veri okumak için kaynak tablo adında düzenli ifade kullanabilirsiniz. Düzenli ifadeler kullanırken dizenin başını ve sonunu eşleştirmek üzere ^ ve $ simgelerini kullanmayın. Daha fazla bilgi için aşağıdaki nota bakın. Not Bir MySQL CDC kaynak tablosu tablo adlarını düzenli ifadeyle eşleştirirken, belirttiğiniz database-name ve table-name değerlerini \\. dizesiyle birleştirerek tam yollu bir düzenli ifade oluşturur. VVR 8.0.1 öncesinde . karakteri kullanılırdı. Bağlayıcı daha sonra bu düzenli ifadeyi kullanarak MySQL veritabanındaki tabloların tam nitelikli adlarını eşleştirir. Örneğin, 'database-name'='db_.' ve 'table-name'='tb_.+' şeklinde ayarlarsanız bağlayıcı, hangi tabloların okunacağını belirlemek için tam nitelikli tablo adlarını eşleştirmek üzere db_.\\.tb_.+ düzenli ifadesini kullanır. VVR 8.0.1 öncesinde düzenli ifade db_.*.tb_.+ idi.
  port	MySQL veritabanı hizmetinin bağlantı noktası numarası.	No	INTEGER	3306	Yok.

• Yalnızca kaynak tablo

  Parametre	Açıklama	Gerekli	Veri türü	Varsayılan değer	Notlar
  server-id	Veritabanı istemcisi için sayısal bir kimlik.	Hayır	STRING	5400 ile 6400 arasında rastgele bir değer oluşturulur.	Bu kimliğin MySQL kümesinde genel olarak benzersiz olması gerekir. Aynı veritabanına bağlanan her iş için farklı bir kimlik ayarlayın. Bu parametre ayrıca 5400-5408 gibi bir kimlik aralığı biçimini de destekler. Artımlı okuma etkinleştirildiğinde eşzamanlı okuma desteklenir. Bu durumda, her eşzamanlı okuyucunun farklı bir kimlik kullanması için bir kimlik aralığı ayarlayın. Daha fazla bilgi için bkz. Sunucu Kimliğini Kullanma.
  scan.incremental.snapshot.enabled	Artımlı anlık görüntülerin etkinleştirilip etkinleştirilmeyeceğini belirtir.	Hayır	BOOLEAN	true	Artımlı anlık görüntüler varsayılan olarak etkindir. Artımlı anlık görüntü, tam veri anlık görüntülerini okumak için yeni bir mekanizmadır. Eski anlık görüntü okuma yöntemine kıyasla artımlı anlık görüntülerin aşağıdakiler dahil olmak üzere birçok avantajı vardır: Kaynak, tam verileri paralel olarak okuyabilir. Kaynak, tam verileri okurken parça düzeyinde kontrol noktalarını destekler. Kaynak, tam verileri okurken genel okuma kilidi (FLUSH TABLES WITH read lock) almayı gerektirmez. Kaynağın eşzamanlı okumayı desteklemesini istiyorsanız her eşzamanlı okuyucunun benzersiz bir sunucu kimliğine ihtiyacı vardır. Bu nedenle server-id bir aralık olmalıdır (örneğin 5400-6400) ve aralığın boyutu eşzamanlılık düzeyinden büyük veya ona eşit olmalıdır. Not Bu yapılandırma öğesi Ververica Runtime (VVR) 11.1 ve sonraki sürümlerde kaldırılmıştır.
  scan.incremental.snapshot.chunk.size	Her bir parçanın satır cinsinden boyutu.	Hayır	INTEGER	8096	Artımlı anlık görüntü okuma etkinleştirildiğinde tablo okuma için birden çok parçaya bölünür. Bir parçanın verileri tamamen okunana kadar bellekte önbelleğe alınır. Her parçada bulunan satır sayısı azaldıkça tabloda toplam parça sayısı artar. Bu durum hata kurtarma granülerliğini azaltsa da bellek yetersizliği (OOM) hatalarına ve daha düşük genel aktarım hızına yol açabilir. Bu nedenle makul bir parça boyutu belirlemek için dengeli bir yaklaşım sergilemeniz gerekir.
  scan.snapshot.fetch.size	Bir tablonun tam verilerini okurken tek seferde çekilecek maksimum kayıt sayısı.	Hayır	INTEGER	1024	Yok.
  scan.startup.mode	Veri tüketimi için başlangıç modu.	Hayır	STRING	initial	Geçerli değerler: initial (varsayılan): İlk başlatmada veya durumsuz bir başlatmada konektör tüm geçmiş verileri tarar ve ardından en son ikili günlük verilerini okur. latest-offset: İlk başlatmada veya durumsuz bir başlatmada konektör geçmiş verileri taramaz. İkili günlüğün sonundan itibaren okumaya başlar; bu, konektör başladıktan sonra yapılan yalnızca en son değişiklikleri okuduğu anlamına gelir. earliest-offset: Konektör geçmiş verileri taramaz. En eski kullanılabilir ikili günlükten başlayarak okumaya başlar. specific-offset: Konektör geçmiş verileri taramaz. Belirli bir ikili günlük ofsetinden başlar. Ofseti hem scan.startup.specific-offset.file hem de scan.startup.specific-offset.pos parametrelerini yapılandırarak veya belirli bir GTID kümesinden başlamak için yalnızca scan.startup.specific-offset.gtid-set parametresini yapılandırarak belirtebilirsiniz. timestamp: Konektör geçmiş verileri taramaz. Belirtilen bir zaman damgasından itibaren ikili günlüğü okumaya başlar. Zaman damgası milisaniye cinsinden scan.startup.timestamp-millis parametresiyle belirtilir. Önemli earliest-offset, specific-offset veya timestamp başlangıç modlarını kullanırken belirtilen ikili günlük tüketim konumu ile işin başlatılma zamanı arasında ilgili tablonun şemasının değişmediğinden emin olun. Bu, şema uyuşmazlıklarından kaynaklanan hataları önler.
  scan.startup.specific-offset.file	specific-offset başlangıç modu kullanılırken başlangıç ofseti için ikili günlük dosya adı.	Hayır	STRING	Yok	Bu parametreyi kullanırken scan.startup.mode parametresini specific-offset olarak ayarlamanız gerekir. Örnek dosya adı biçimi: mysql-bin.000003.
  scan.startup.specific-offset.pos	specific-offset başlangıç modu kullanılırken belirtilen ikili günlük dosyasındaki başlangıç ofseti.	Hayır	INTEGER	Yok	Bu parametreyi kullanırken scan.startup.mode parametresini specific-offset olarak ayarlamanız gerekir.
  scan.startup.specific-offset.gtid-set	specific-offset başlangıç modu kullanılırken başlangıç ofseti için GTID kümesi.	Hayır	STRING	Yok	Bu parametreyi kullanırken scan.startup.mode parametresini specific-offset olarak ayarlamanız gerekir. Örnek GTID kümesi biçimi: 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19.
  scan.startup.timestamp-millis	timestamp başlangıç modu kullanılırken başlangıç ofseti için milisaniye cinsinden zaman damgası.	Hayır	LONG	Yok	Bu parametreyi kullanırken scan.startup.mode parametresini timestamp olarak ayarlamanız gerekir. Zaman damgası birimi milisaniyedir. Önemli Bir zaman belirttiğinizde MySQL CDC, her ikili günlük dosyasının ilk olayını okuyarak zaman damgasını belirlemeye çalışır. Ardından belirtilen zamana karşılık gelen ikili günlük dosyasını bulur. Belirtilen zaman damgasına karşılık gelen ikili günlük dosyasının veritabanından temizlenmediğinden ve okunabilir olduğundan emin olun.
  server-time-zone	Veritabanı tarafından kullanılan oturum saat dilimi.	Hayır	STRING	Bu parametreyi belirtmezseniz sistem, Flink iş çalışma zamanının ortam saat dilimini veritabanı sunucusu saat dilimi olarak kullanır. Bu, seçtiğiniz alanın saat dilimidir.	Örnek: Asia/Shanghai. Bu parametre, MySQL'deki TIMESTAMP türünün STRING türüne nasıl dönüştürüleceğini kontrol eder. Daha fazla bilgi için bkz. Debezium temporal values.
  debezium.min.row.count.to.stream.results	Bir tablodaki satır sayısı bu değerden büyük olduğunda toplu okuma modu kullanılır.	Hayır	INTEGER	1000	Flink, bir MySQL kaynak tablosundan verileri aşağıdaki yollardan biriyle okur: Tam okuma: Tablonun tüm verilerini doğrudan belleğe okur. Bu yöntem hızlıdır ancak karşılık gelen miktarda bellek tüketir. Kaynak tablo çok büyükse OOM hatası riski vardır. Toplu okuma: Tüm veriler okunana kadar her topluda belirli sayıda satır olacak şekilde verileri birden çok toplu halinde okur. Bu yöntem büyük tabloları okurken OOM riskini önler ancak nispeten yavaştır.
  connect.timeout	MySQL veritabanı sunucusuna bağlantı kurulurken yeniden denemeden önce beklenen maksimum zaman aşımı süresi.	Hayır	DURATION	30s	Yok.
  connect.max-retries	MySQL veritabanı hizmetine başarısız bir bağlantıdan sonra yapılacak maksimum yeniden deneme sayısı.	Hayır	INTEGER	3	Yok.
  connection.pool.size	Veritabanı bağlantı havuzunun boyutu.	Hayır	INTEGER	20	Veritabanı bağlantı havuzu, bağlantıları yeniden kullanarak veritabanı bağlantısı sayısını azaltır.
  jdbc.properties.*	JDBC URL'sindeki özel bağlantı parametreleri.	Hayır	STRING	Yok	Özel bağlantı parametreleri iletebilirsiniz. Örneğin SSL protokolünü kullanmamak için 'jdbc.properties.useSSL' = 'false' şeklinde yapılandırma yapabilirsiniz. Desteklenen bağlantı parametreleri hakkında daha fazla bilgi için bkz. MySQL Configuration Properties.
  debezium.*	Debezium'un ikili günlükleri okuması için özel parametreler.	Hayır	STRING	Yok	Özel Debezium parametreleri iletebilirsiniz. Örneğin ayrıştırma hataları için işleme mantığını belirtmek amacıyla 'debezium.event.deserialization.failure.handling.mode'='ignore' kullanabilirsiniz. Uyarı Debezium parametrelerini keyfi olarak değiştirmeyin. Bu, konektörün verileri yanlış okumasına neden olabilir. Örneğin debezium.binlog.buffer.size parametresinin yapılandırılmasına izin verilmez.
  heartbeat.interval	Kaynağın kalp atışı olaylarını kullanarak ikili günlük ofsetini ilerletme aralığı.	Hayır	DURATION	30s	Kalp atışı olayları, kaynaktaki ikili günlük ofsetini ilerletmek için kullanılır. Bu durum, MySQL'de nadiren güncellenen tablolar için oldukça faydalıdır. Bu tür tablolar için ikili günlük ofseti otomatik olarak ilerleyemez. Kalp atışı olayları ikili günlük ofsetini ileriye taşıyarak süresi dolmuş bir ikili günlük ofsetinden kaynaklanan sorunları önler. Süresi dolmuş bir ikili günlük ofseti işin başarısız olmasına ve kurtarılamamasına neden olabilir; bu da durumsuz bir yeniden başlatma gerektirir.
  scan.incremental.snapshot.chunk.key-column	Anlık görüntü aşamasında parçalama için bölme sütunu olarak kullanılacak bir sütun belirtir.	Notlar sütununa bakın.	STRING	Yok	Birincil anahtarı olmayan tablolar için gereklidir. Seçilen sütun null olmayan bir türde (NOT NULL) olmalıdır. Birincil anahtarı olan tablolar için isteğe bağlıdır. Birincil anahtardan yalnızca bir sütun seçilebilir.
  rds.region-id	Alibaba Cloud ApsaraDB RDS for MySQL örneğinin bölge kimliği.	OSS'ten arşivlenmiş günlükleri okuma özelliği kullanılırken gereklidir.	STRING	Yok	Bölge kimlikleri hakkında daha fazla bilgi için bkz. Bölgeler ve alanlar. Önemli MySQL CDC için GTID dizesi rastgele oluşturulduğundan ve ikili günlük dosya ofsetleri gibi monoton artmadığından bir dosyada GTID bulmak OSS'ten tüm arşivlenmiş günlüklerin indirilip ayrıştırılmasını gerektirir. Bu işlem çok yoğun kaynak tüketir ve zaman alır; bu da GTID ofsetlerine dayalı özelliklerin uygulanmasını imkansız kılar. Bu nedenle OSS arşivlenmiş günlük özelliği yalnızca belirtilen bir zaman damgasından veya belirtilen bir ikili günlük dosya ofsetinden başlamayı destekler. Belirtilen bir GTID'den başlamayı veya arşivlenmiş günlüklerde birincil/ikincil geçişleri içeren senaryoları desteklemez çünkü MySQL birincil/ikincil geçişleri GTID'lere dayanır. Bu özelliği kullanmadan önce dikkatlice değerlendirin.
  rds.access-key-id	Alibaba Cloud ApsaraDB RDS for MySQL hesabının AccessKey kimliği.	OSS'ten arşivlenmiş günlükleri okuma özelliği kullanılırken gereklidir.	STRING	Yok	Daha fazla bilgi için bkz. AccessKey ID ve AccessKey gizli anahtarını nasıl görüntülerim?. Önemli AccessKey bilgilerinizin sızmasını önlemek için AccessKey kimliğini belirtmek üzere gizli anahtar yönetimi özelliğini kullanın. Daha fazla bilgi için bkz. Değişkenleri yönetme.
  rds.access-key-secret	Alibaba Cloud ApsaraDB RDS for MySQL hesabının AccessKey gizli anahtarı.	OSS'ten arşivlenmiş günlükleri okuma özelliği kullanılırken gereklidir.	STRING	Yok	Daha fazla bilgi için bkz. AccessKey ID ve AccessKey gizli anahtarını nasıl görüntülerim? Önemli AccessKey bilgilerinizin sızmasını önlemek için AccessKey gizli anahtarını belirtmek üzere gizli anahtar yönetimi özelliğini kullanın. Daha fazla bilgi için bkz. Değişkenleri yönetme.
  rds.db-instance-id	Alibaba Cloud ApsaraDB RDS for MySQL örneğinin kimliği.	OSS'ten arşivlenmiş günlükleri okuma özelliği kullanılırken gereklidir.	STRING	Yok	Yok.
  rds.main-db-id	Alibaba Cloud ApsaraDB RDS for MySQL örneğinin birincil veritabanı numarası.	Hayır	STRING	Yok	Birincil veritabanı numarasının nasıl elde edileceği hakkında daha fazla bilgi için bkz. ApsaraDB RDS for MySQL günlük yedekleme. Yalnızca VVR 8.0.7 ve sonraki sürümlerde desteklenir. Not Bu parametre belirtilmezse VVR 11.7 ve sonraki sürümler, ApsaraDB RDS for MySQL bağlantı bilgilerine göre birincil veritabanı numarasını otomatik olarak sorgular.
  rds.download.timeout	OSS'ten tek bir arşivlenmiş günlüğü indirmek için zaman aşımı süresi.	Hayır	DURATION	60s	Yok.
  rds.endpoint	OSS ikili günlük bilgilerini almak için hizmet uç noktası.	Hayır	STRING	Yok	Geçerli değerler hakkında daha fazla bilgi için bkz. Uç noktalar. Yalnızca VVR 8.0.8 ve sonraki sürümlerde desteklenir.
  scan.incremental.close-idle-reader.enabled	Anlık görüntü tamamlandıktan sonra boşta kalan okuyucuların kapatılıp kapatılmayacağını belirtir.	Hayır	BOOLEAN	false	Yalnızca VVR 8.0.1 ve sonraki sürümlerde desteklenir. Bu yapılandırmanın etkili olması için execution.checkpointing.checkpoints-after-tasks-finish.enabled parametresini true olarak ayarlamanız gerekir.
  scan.read-changelog-as-append-only.enabled	Değişiklik günlüğü veri akışının yalnızca ekleme yapılabilen bir veri akışına dönüştürülüp dönüştürülmeyeceğini belirtir.	Hayır	BOOLEAN	false	Geçerli değerler: true: INSERT, DELETE, UPDATE_BEFORE ve UPDATE_AFTER dahil tüm mesaj türleri INSERT mesajlarına dönüştürülür. Bu seçeneği yalnızca yukarı akış tablosundaki silme mesajlarını kaydetmeniz gerektiğinde gibi özel senaryolarda etkinleştirin. false (varsayılan): Tüm mesaj türleri olduğu gibi aşağı akışa gönderilir. Not Yalnızca VVR 8.0.8 ve sonraki sürümlerde desteklenir.
  scan.only.deserialize.captured.tables.changelog.enabled	Artımlı aşamada yalnızca belirtilen tabloların değişiklik olaylarının ayrıştırılıp ayrıştırılmayacağını belirtir.	Hayır	BOOLEAN	Varsayılan değer VVR 8.x sürümlerinde false'tur. Varsayılan değer VVR 11.1 ve sonraki sürümlerde true'dur.	Geçerli değerler: true: İkili günlük okumayı hızlandırmak için yalnızca hedef tabloların değişiklik verilerini ayrıştırır. false (varsayılan): Tüm tabloların değişiklik verilerini ayrıştırır. Not Yalnızca VVR 8.0.7 ve sonraki sürümlerde desteklenir. VVR 8.0.8 veya önceki sürümleri kullanırken parametre adını debezium.scan.only.deserialize.captured.tables.changelog.enable olarak değiştirmeniz gerekir.
  scan.parse.online.schema.changes.enabled	Artımlı aşamada RDS kilitsiz değişiklik DDL olaylarını ayrıştırmayı denemenin gerekip gerekmediğini belirtir.	Hayır	BOOLEAN	false	Geçerli değerler: true: RDS kilitsiz değişiklik DDL olaylarını ayrıştırır. false (varsayılan): RDS kilitsiz değişiklik DDL olaylarını ayrıştırmaz. Bu deneysel bir özelliktir. Çevrimiçi kilitsiz bir değişiklik gerçekleştirmeden önce kurtarma amacıyla Flink işinin anlık görüntüsünü alın. Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
  scan.incremental.snapshot.backfill.skip	Anlık görüntü okuma aşamasında geri doldurmanın atlanıp atlanmayacağını belirtir.	Hayır	BOOLEAN	false	Geçerli değerler: true: Anlık görüntü okuma aşamasında geri doldurmayı atlar. false (varsayılan): Anlık görüntü okuma aşamasında 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 aşamasını kapsamaz. Geri doldurma atlandığında her parçanın anlık görüntü sorgusu o andaki en güncel tablo verilerini okur; bir parça okunduktan sonra o parçada gerçekleşen güncellemeler tam okuma aşamasında birleştirilmez ve artımlı aşamaya girildikten sonra Binlog'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; okuyucu chunk80'e ilerledikten sonra chunk5 güncellenirse güncelleme artımlı aşamada Binlog'dan daha sonra uygulanır. Önemli Etkinleştirildiğinde bir parçanın taraması sırasında veya sonrasında meydana gelen değişiklikler artımlı aşamada Binlog'dan teslim edilir ve yinelenen kayıtlara neden olabilir. Yalnızca en az bir kez semantiği garanti edilir. Bunu yalnızca aşağı akış havuzu birincil anahtara göre yedimsiz yazmaları desteklediğinde etkinleştirin. Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
  scan.incremental.snapshot.unbounded-chunk-first.enabled	Anlık görüntü okuma aşamasında sınırsız parçaların öncelikli olarak dağıtılıp dağıtılmayacağını belirtir.	Hayır	BOOELEAN	false	Geçerli değerler: true: Anlık görüntü okuma aşamasında sınırsız parçaları öncelikli olarak dağıtır. false (varsayılan): Anlık görüntü okuma aşamasında sınırsız parçaları öncelikli olarak dağıtmaz. Bu deneysel bir özelliktir. Etkinleştirilmesi, anlık görüntü aşamasında son parçayı senkronize ederken TaskManager'da OOM hatası riskini azaltabilir. Bu parametreyi işin ilk başlatmasından önce ekleyin. Not Yalnızca VVR 11.1 ve sonraki sürümlerde desteklenir.
  binlog.session.network.timeout	İkili günlük bağlantısı için ağ okuma/yazma zaman aşımı.	Hayır	DURATION	10m	0s olarak ayarlanırsa MySQL sunucusunun varsayılan zaman aşımı kullanılır. Not Yalnızca VVR 11.5 ve sonraki sürümlerde desteklenir.
  scan.rate-limit.records-per-second	Kaynağın saniyede gönderdiği maksimum kayıt sayısını sınırlar.	Hayır	LONG	Yok	Bu parametre veri okumanın sınırlandırılması gereken senaryolar için uygundur. Bu sınır hem tam hem de artımlı aşamalarda etkilidir. Kaynağın numRecordsOutPerSecond ölçümü, tüm veri akışı tarafından saniyede çıktılanan kayıt sayısını yansıtır. Bu ölçüme göre parametreyi ayarlayabilirsiniz. Tam veri okuma aşamasında genellikle her topluda okunan satır sayısını azaltmanız gerekir. scan.incremental.snapshot.chunk.size parametresinin değerini düşürebilirsiniz. Not Yalnızca VVR 11.5 ve sonraki sürümlerde desteklenir.
  scan.binlog.tolerate.gtid-holes	Bu parametrenin etkinleştirilmesi GTID dizisindeki boşlukları yok sayarak işin kesintili olayları atlayıp çalışmaya devam etmesini sağlar.	Hayır	BOOLEAN	false	Bu parametreyi etkinleştirmeden önce işin başlangıç ofsetinin süresinin dolmadığından emin olmalısınız. İş temizlenmiş veya süresi dolmuş bir GTID ofsetinden başlarsa motor eksik günlükleri sessizce atlayarak veri kaybına yol açar. Not Bu parametre yalnızca VVR 11.6 ve sonraki sürümlerde desteklenir.
  scan.filter-unchanged-row.enabled	SQL'de kullanılmayan alanların değişiklik olaylarının filtrelenip filtrelenmeyeceğini belirtir.	Hayır	BOOLEAN	false	Bu parametreyi true olarak ayarlarsanız bir CDC kaynak tablosu bir UPDATE olayı aldığında yalnızca SQL ifadesinde başvurulan sütunları karşılaştırır. Başvurulan sütunların değerleri değişmezse olay aşağı akışa gönderilmez. Bu parametre çoklu akış JOIN senaryolarında etkili olur. Yukarı akış tablosu SQL ifadesinde başvurulan bir alanı güncellerse gereksiz DELETE ve INSERT olayları oluşmaz. Bu durum aşağı akış operatörlerinde veri çoğalmasını ve sonuç titremesini azaltır. Not Bu parametre yalnızca VVR 11.8 ve sonraki sürümlerde desteklenir.
  scan.projection-pushdown.enabled	Projeksiyon itelemenin etkinleştirilip etkinleştirilmeyeceğini belirtir.	Hayır	BOOLEAN	false	Bu parametreyi true olarak ayarlarsanız bir CDC kaynak tablosu yalnızca SQL ifadesinde başvurulan sütunları ayrıştırır ve iletir. Başvurulmayan sütunlar serileştirilmez veya ağ üzerinden iletilmez. Bu durum G/Ç ve hesaplama yükünü azaltır. Bu parametre scan.filter-unchanged-row.enabled ile birlikte kullanıldığında daha iyi çalışır. Projeksiyon iteleme önce veri hacmini azaltır ardından değişmeyen satırlar filtrelenerek aşağı akış operatörlerindeki işleme yükü daha da hafifletilir. Not Bu parametre yalnızca VVR 11.8 ve sonraki sürümlerde desteklenir.

• Boyut tablosuna özel parametreler

  Parametre	Açıklama	Gerekli	Veri türü	Varsayılan değer	Notlar
  url	MySQL JDBC URL'si.	Hayır	STRING	Yok	URL biçimi şöyledir: jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	Başarısız bir veri okuma işleminden sonra gerçekleştirilecek maksimum yeniden deneme sayısı.	Hayır	INTEGER	3	Yalnızca VVR 6.0.7 ve sonraki sürümlerde desteklenir.
  lookup.cache.strategy	Önbellek ilkesi.	Hayır	STRING	Yok	Desteklenen önbellek ilkeleri None, LRU ve ALL'dir. Değerler hakkında daha fazla bilgi için bkz. Boyut tablosu JOIN deyimleri. Not LRU önbellek ilkesini kullandığınızda lookup.cache.max-rows parametresini de yapılandırmanız gerekir.
  lookup.cache.max-rows	Maksimum önbelleğe alınmış satır sayısı.	Hayır	INTEGER	100000	LRU önbellek ilkesini seçtiyseniz önbellek boyutunu ayarlamanız gerekir. ALL önbellek ilkesini seçtiyseniz önbellek boyutunu ayarlamanıza gerek yoktur.
  lookup.cache.ttl	Önbellek yaşam süresi (TTL).	Hayır	DURATION	10 s	lookup.cache.ttl yapılandırması lookup.cache.strategy değerine bağlıdır: lookup.cache.strategy değeri None olarak ayarlanmışsa lookup.cache.ttl öğesini yapılandırmanıza gerek yoktur. Bu durumda önbelleğin zaman aşımı olmaz. lookup.cache.strategy değeri LRU olarak ayarlanmışsa lookup.cache.ttl önbellek TTL değeridir. Varsayılan olarak önbelleğin süresi dolmaz. lookup.cache.strategy değeri ALL olarak ayarlanmışsa lookup.cache.ttl önbellek yükleme süresidir. Varsayılan olarak önbellek yeniden yüklenmez. 1min veya 10s gibi bir zaman biçimi kullanın.
  lookup.max-join-rows	Birincil tablodan bir kaydın boyut tablosundaki kayıtlarla eşleşmesi durumunda döndürülen maksimum sonuç sayısı.	Hayır	INTEGER	1024	Yok.
  lookup.filter-push-down.enabled	Boyut tablosu için filtre aşağı itmenin etkinleştirilip etkinleştirilmeyeceğini belirtir.	Hayır	BOOLEAN	false	Geçerli değerler: true: Boyut tablosu için filtre aşağı itmeyi etkinleştirir. MySQL veritabanı tablosundan veri yüklerken boyut tablosu, SQL işinde belirtilen koşullara göre verileri önceden filtreler. false (varsayılan): Boyut tablosu için filtre aşağı itmeyi devre dışı bırakır. MySQL veritabanı tablosundan veri yüklerken boyut tablosu tüm verileri yükler. Not Yalnızca VVR 8.0.7 ve sonraki sürümlerde desteklenir. Önemli Boyut tablosu aşağı itme özelliği yalnızca bir Flink tablosu boyut tablosu olarak kullanıldığında etkinleştirilmelidir. MySQL source tabloları filtre aşağı itmenin etkinleştirilmesini desteklemez. Bir Flink tablosu hem source tablosu hem de boyut tablosu olarak kullanılıyorsa ve boyut tablosu için filtre aşağı itme etkinleştirilmişse source tablosu için bu yapılandırma öğesini SQL İpuçları kullanarak açıkça false olarak ayarlamanız gerekir. Aksi takdirde iş anormal şekilde çalışabilir.
  lookup.cache.cache-empty	Boş sorgu sonuçlarının önbelleğe alınıp alınmayacağını belirtir.	Hayır	BOOLEAN	true	Yalnızca önbellek ilkesi LRU olduğunda etkili olur. Geçerli değerler: true (varsayılan): Bir sorgu veri döndürmezse boş bir sonuç önbelleğe alınır ve sonraki sorgular için doğrudan döndürülür. false: Bir sorgu veri döndürmezse hiçbir sonuç önbelleğe alınmaz ve sonraki sorgu verileri tekrar arar. Not Bu parametre yalnızca VVR 11.9.0 ve sonraki sürümlerde desteklenir.

• Yalnızca sink tabloları için

  Parametre	Açıklama	Gerekli	Veri türü	Varsayılan değer	Notlar
  url	MySQL JDBC URL'si.	Hayır	STRING	Yok	URL biçimi şöyledir: jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Başarısız bir veri yazma işleminden sonra gerçekleştirilecek maksimum yeniden deneme sayısı.	Hayır	INTEGER	3	Yok.
  sink.buffer-flush.batch-size	Tek bir toplu iş yazmasındaki satır sayısı.	Hayır	INTEGER	4096	Yok.
  sink.buffer-flush.max-rows	Bellekte önbelleğe alınan veri satırı sayısı.	Hayır	INTEGER	10000	Bu parametre yalnızca bir birincil anahtar belirtildikten sonra etkili olur.
  sink.buffer-flush.interval	Önbelleği temizleme aralığı. Belirtilen bekleme süresi sonunda önbellekteki veriler çıkış koşullarını karşılamazsa sistem önbellekteki tüm verileri otomatik olarak çıkarır.	Hayır	DURATION	1s	Yok.
  sink.ignore-delete	Veri SİLME işlemlerinin yoksayılıp yoksayılmayacağını belirtir.	Hayır	BOOLEAN	false	Flink SQL tarafından oluşturulan akış delete veya update-before kayıtları içeriyorsa birden fazla çıkış görevi aynı tablonun farklı alanlarını aynı anda güncellediğinde veri tutarsızlığı oluşabilir. Örneğin bir kayıt silindikten sonra başka bir görev yalnızca bazı alanları günceller. Güncellenmemiş alanlar null veya varsayılan değerlere dönüşerek veri hatalarına neden olur. sink.ignore-delete değerini true olarak ayarlayarak yukarı akıştaki DELETE ve UPDATE_BEFORE işlemlerini yoksayabilir ve bu tür sorunlardan kaçınabilirsiniz. Not UPDATE_BEFORE, bir güncelleme işlemindeki eski değeri "geri çekmek" için kullanılan Flink'in geri çekme mekanizmasının bir parçasıdır. ignoreDelete = true olduğunda tüm DELETE ve UPDATE_BEFORE türündeki kayıtlar atlanır. Yalnızca INSERT ve UPDATE_AFTER kayıtları işlenir.
  sink.ignore-delete-mode	DELETE işlemleri yoksayıldıktan sonra silme türündeki kayıtları işleme stratejisi.	Hayır	STRING	ALL	Geçerli değerler: ALL: Hem -D hem de -U kayıtlarını yoksayar. REAL_DELETE: Yalnızca -D kayıtlarını yoksayar. UPDATE_BEFORE: Yalnızca -U kayıtlarını yoksayar. Not Bu seçenek yalnızca Realtime Compute motoru VVR 11.8 ve sonraki sürümlerde desteklenir. Yalnızca sink.ignore-delete=true olduğunda etkilidir. Tek başına yapılandırılması hataya neden olur.
  sink.ignore-null-when-update	Verileri güncellerken gelen veri alanı değeri null ise ilgili alanın null olarak güncellenip güncellenmeyeceğini veya o alan için güncellemenin atlanıp atlanmayacağını belirtir.	Hayır	BOOLEAN	false	Geçerli değerler: true: Alanı güncellemez. Bu parametre yalnızca Flink tablosu için bir birincil anahtar ayarlandığında true olarak ayarlanabilir. true olarak ayarlandığında: VVR 8.0.6 ve önceki sürümler için sink tablosu toplu iş yazmayı desteklemez. VVR 8.0.7 ve sonraki sürümler için sink tablosu toplu iş yazmayı destekler. Toplu iş yazma, yazma verimliliğini ve genel aktarım hızını önemli ölçüde artırabilir ancak veri gecikmesine ve OOM hataları riskine neden olur. Bu nedenle iş senaryonza göre bir denge kurmanız gerekir. false: Alanı null olarak günceller. Not Bu parametre yalnızca VVR 8.0.5 ve sonraki sürümlerde desteklenir.
  sink.force-batch-on-non-primary-table	Birincil anahtarı olmayan bir MySQL sink tablosuna veri yazılırken kayıtların zorla arabelleğe alınıp JDBC Batch kullanılarak toplu iş olarak eklenip eklenmeyeceğini belirtir.	Hayır	BOOLEAN	false	Bu parametre yalnızca birincil anahtarı olmayan tablolarda etkili olur. Bu parametreyi true olarak ayarlarsanız DELETE ve UPDATE_BEFORE olayları sessizce atılır ve INSERT ile UPDATE_AFTER olayları INSERT deyimleri olarak toplu iş halinde yazılır. Verilerin yalnızca ekleme amaçlı olması veya önceki semantiğin sizin için kabul edilebilir olması durumunda bu parametreyi etkinleştirmenizi öneririz. Not Bu parametre yalnızca VVR 11.8 ve sonraki sürümlerde desteklenir. Verilerin yalnızca ekleme amaçlı olması veya önceki semantiğin sizin için kabul edilebilir olması durumunda bu parametreyi etkinleştirin. Aksi takdirde veriler tutarsız hale gelebilir.

Tür eşleme

• CDC kaynak tabloları

  MySQL CDC alan türü	Flink alan türü
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  TINYINT UNSIGNED ZEROFILL
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  SMALLINT UNSIGNED ZEROFILL
  BIGINT	BIGINT
  INT UNSIGNED
  INT UNSIGNED ZEROFILL
  MEDIUMINT UNSIGNED
  MEDIUMINT UNSIGNED ZEROFILL
  BIGINT UNSIGNED	DECIMAL(20, 0)
  BIGINT UNSIGNED ZEROFILL
  SERIAL
  FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
  DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
  DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
  REAL [UNSIGNED] [ZEROFILL]
  NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
  DECIMAL(p, s) [UNSIGNED] [ZEROFILL]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  Önemli

  MySQL'de 0 ve 1 dışındaki değerleri saklamak için TINYINT(1) türünü kullanmayın. property-version=0 olduğunda, MySQL CDC kaynak tablosu TINYINT(1) türünü varsayılan olarak Flink'teki BOOLEAN türüne eşler. Bu durum veri tutarsızlıklarına yol açabilir. TINYINT(1) türünü 0 ve 1 dışındaki değerleri saklamak amacıyla kullanmak istiyorsanız catalog.table.treat-tinyint1-as-boolean yapılandırma parametresine bakın.

• Boyut tabloları ve havuz tabloları

  MySQL alan türü	Flink alan türü
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) Not p <= 38 olmalıdır.
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES Önemli Flink yalnızca boyutu 2.147.483.647 (2^31 - 1) bayt veya daha küçük olan MySQL BLOB türündeki kayıtları destekler.
  BLOB
  MEDIUMBLOB
  LONGBLOB

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Debezium configuration
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Enable parsing filter
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Parse only the change events of specified tables.
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

İş aktarım hızını optimize etmek için veri gecikmesini teşhis etme

Artımlı aşamada veri gecikmesi yaşıyorsanız sorunu aşağıdaki adımları izleyerek analiz edin:

1. Genel Bakış sayfasındaki currentFetchEventTimeLag ve currentEmitEventTimeLag ölçümlerini kontrol edin. currentFetchEventTimeLag ölçümü, ikili günlükten veri okuma gecikmesini temsil eder. currentEmitEventTimeLag ölçümü ise işle ilgili tablolar için ikili günlükten veri okuma gecikmesini temsil eder.

   Senaryo	Açıklama
   currentFetchEventTimeLag düşük, ancak currentEmitEventTimeLag yüksek ve nadiren güncelleniyor.	Düşük currentFetchEventTimeLag değeri, veritabanından ikili günlüğün çekilmesinin verimli olduğunu gösterir. Ancak ikili günlük, işin okuması gereken tablolar için çok az veri içerir. Bu nedenle currentEmitEventTimeLag nadiren güncellenir. Bu beklenen bir davranıştır.
   currentFetchEventTimeLag ve currentEmitEventTimeLag değerlerinin her ikisi de yüksektir.	Bu durum, kaynak tablonun okuma performansının düşük olduğunu gösterir. Optimizasyon için bu bölümdeki sonraki adımlara devam edebilirsiniz.

2. Arka basınç, kaynağın aşağı akış operatörlerine veri gönderme hızını düşürebilir. sourceIdleTime değerinin periyodik olarak arttığını ve hem currentFetchEventTimeLag hem de currentEmitEventTimeLag değerlerinin sürekli büyüdüğünü gözlemleyebilirsiniz. Bu sorunu çözmek için arka basıncın kaynağı olan düğümün paralellliğini artırın.

3. Yetersiz CPU veya bellek kaynağı olup olmadığını belirlemek için CPU sayfasındaki TM CPU Kullanımı ölçümünü ve JVM sayfasındaki TM GC Süresi ölçümünü kontrol edin. Okuma performansını optimize etmek için iş kaynaklarını artırabilirsiniz. Aktarım hızını artırmak için mini-toplu iş parametrelerini de etkinleştirebilirsiniz. Daha fazla bilgi için bkz. Yüksek Performanslı Flink SQL Optimizasyon Teknikleri.

4. İşte büyük bir duruma sahip bir SinkUpsertMaterializer operatörü varsa bu durum okuma performansını etkileyebilir. İş paralelliğini artırmayı veya SinkUpsertMaterializer operatöründen kaçınmayı düşünün. Daha fazla bilgi için bkz. SinkUpsertMaterializer kullanımından kaçının. Mevcut bir işten SinkUpsertMaterializer operatörünü kaldırmak durumsuz yeniden başlatma gerektirir. Bunun nedeni iş topolojisinin değişmesidir; mevcut durumdan başlamak işin başarısız olmasına veya veri kaybına neden olabilir.

Parça parametre optimizasyonu

Parça verileri ve meta verileri bellekte saklanır ve bu durum bazen bellek yetersizliği (OOM) hatalarına yol açabilir. OOM hatasının hangi bileşende oluştuğuna bağlı olarak parametreleri ayarlayabilirsiniz:

• JobManager

  JobManager tüm parçaların meta verilerini saklar. Aşırı sayıda parça OOM hatasına neden olabilir. Bu sorunu çözmek için parça sayısını azaltmak amacıyla scan.incremental.snapshot.chunk.size değerini artırın. Alternatif olarak, çalışma zamanı yapılandırmanızda jobmanager.memory.heap.size ayarını yaparak JobManager yığın belleğini artırabilirsiniz. Daha fazla bilgi için bkz. Flink Parameter Configuration.

• TaskManager

  • TaskManager her parçanın verilerini okur. Bir parça çok fazla satır içeriyorsa OOM hatası oluşabilir. Bu sorunu çözmek için parça başına satır sayısını azaltmak amacıyla scan.incremental.snapshot.chunk.size değerini azaltın. Çalışma zamanı yapılandırmanızda TaskManager Memory değerini artırarak TaskManager yığın belleğini de büyütebilirsiniz.

  • VVR 8.0.8 ve önceki sürümlerde son parça büyük miktarda veri içerebilir ve bu durum TaskManager'da OOM hatasına neden olabilir. Bu sorundan kaçınmak için VVR 8.0.9 veya sonraki bir sürüme yükseltmenizi öneririz.

  • Bileşik birincil anahtara sahip bir MySQL CDC kaynak tablosu için parçalar varsayılan olarak anahtarın ilk sütununa göre bölünür. Veriler önemli ölçüde eğimliyse ve o sütunda aynı değere sahip birçok satır varsa bu değere ait parça çok büyük hale gelebilir ve TaskManager'da OOM hatasına neden olabilir. Bölme için birincil anahtardan farklı bir sütun belirtmek üzere scan.incremental.snapshot.chunk.key-column ayarını yapabilirsiniz.

SET 'table.optimizer.source-merge.enabled' = 'true';

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    'rds.region-id' = 'cn-beijing',
    'rds.access-key-id' = 'your_access_key_id',
    'rds.access-key-secret' = 'your_access_key_secret',
    'rds.db-instance-id' = 'rm-xxxxxxxx',  // The ID of the database instance.
    'rds.main-db-id' = '12345678', // The ID of the primary database.
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)