Realtime Compute for Apache Flink:Kafka connector

Genel bakış

Önkoşullar

Notlar

Apache Flink ve Apache Kafka'daki bilinen tasarım sınırlamaları nedeniyle transactional yazmalar önerilmez. sink.delivery-guarantee = 'exactly-once' ayarlandığında Kafka bağlayıcısı transactional yazmaları etkinleştirir ve aşağıdaki bilinen sorunlar ortaya çıkar:

• Her checkpoint yeni bir Transaction ID üretir. Checkpoint aralığı çok kısa olursa oluşan Transaction ID yoğunluğu Kafka küme koordinatörünün bellek yetersizliğine yol açarak küme kararlılığını tehlikeye atabilir.

• Her transaction yeni bir Producer örneği oluşturur. Çok fazla transaction eşzamanlı olarak commit edilirse TaskManager bellek yetersizliğine girebilir ve Apache Flink işinin kararsız hale gelmesine neden olabilir.

• Birden fazla Apache Flink işi aynı sink.transactional-id-prefix değerini kullanırsa üretilen Transaction ID'ler çakışabilir. Bir işteki yazma işlemi başarısız olduğunda bu durum bir Apache Kafka bölümünün Log Start Offset (LSO) değerinin ilerlemesini engelleyebilir. Bu da o bölümün tüm tüketicilerini etkiler.

Exactly-once semantiğine ihtiyacınız varsa idempotence sağlamak amacıyla verileri birincil anahtarlı bir tabloya yazmak için Upsert Kafka connector kullanın. Transactional yazmaları kullanmak zorundaysanız bkz. Exactly-once semantics usage notes.

Ağ bağlantısı sorunlarını giderme

Bir Realtime Compute for Apache Flink işi başlatılamadığında出现的 Timed out waiting for a node assignment hatası genellikle Realtime Compute for Apache Flink ile Kafka kümesi arasında bir ağ bağlantısı sorunu olduğunu gösterir.

SQL

Kafka bağlayıcısını SQL işlerinde kaynak veya hedef tablo olarak kullanın.

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

Meta veri sütunları

CREATE TABLE kafka_source (
  -- Read the message topic as the `record_topic` column
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` column
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the message offset as the `record_offset` column
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Write the timestamp from the `ts` column as the ProducerRecord's timestamp to Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

Bağlayıcı seçenekleri

• Genel

  Seçenek	Açıklama	Tür	Gerekli	Varsayılan	Açıklamalar
  connector	Bağlayıcı türü.	String	Evet	–	Değer kafka olmalıdır.
  properties.bootstrap.servers	Kafka broker adreslerinin listesi.	String	Evet	–	Format: host1:port1,host2:port2,.... Adresleri virgül (,) ile ayırın.
  properties.*	Kafka istemcisi için ek özellikler.	String	Hayır	–	Özellik anahtarları resmi Apache Kafka belgelerinde Producer Configs ve Consumer Configs için tanımlanan geçerli seçenekler olmalıdır. Realtime Compute for Apache Flink properties. önekini kaldırır ve kalan anahtar-değer çiftlerini alttaki Kafka istemcisine iletir. Örneğin, otomatik konu oluşturmayı devre dışı bırakmak için 'properties.allow.auto.create.topics' = 'false' ayarlayabilirsiniz. Kafka bağlayıcısı bu seçeneklerin üzerine yazar, bu nedenle bunları bu şekilde yapılandıramazsınız: key.deserializer value.deserializer
  format	Bir Kafka mesajının değerini serileştirmek ve seri durumdan çıkarmak için kullanılan format.	String	Hayır	–	Desteklenen formatlar: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Not Daha fazla bilgi için bkz. Format options.
  key.format	Bir Kafka mesajının anahtarını serileştirmek ve seri durumdan çıkarmak için kullanılan format.	String	Hayır	–	Desteklenen formatlar: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Not Bu yapılandırmayı kullandığınızda key.options gereklidir.
  key.fields	Kafka mesaj anahtarı olarak kullanılacak tablo şemasından alanlar.	String	Hayır	–	Birden fazla alan adını noktalı virgül (;) ile ayırın. Örneğin, 'field1;field2'.
  key.fields-prefix	Tüm anahtar alanları için değer alanlarıyla ad çakışmalarını önlemek amacıyla özel bir önek.	String	Hayır	–	Bu önek anahtar ve değer alanlarını ayırt etmek için kullanılır. Anahtarı serileştirmeden önce veya seri durumdan çıkardıktan sonra kaldırılır. Not Bu seçeneği kullanırsanız value.fields-include değeri EXCEPT_KEY olarak ayarlanmalıdır.
  value.format	Bir Kafka mesajının değerini serileştirmek ve seri durumdan çıkarmak için kullanılan format.	String	Hayır	–	Bu yapılandırma format ile eşdeğerdir. format veya value.format seçeneklerinden yalnızca birini ayarlayabilirsiniz. Her ikisi de yapılandırılırsa value.format format seçeneğinin üzerine yazar.
  value.fields-include	Anahtar alanların değer formatına dahil edilip edilmeyeceğini tanımlar.	String	Hayır	ALL	Geçerli değerler: ALL: Kafka mesaj değeri tüm tablo sütunlarını içerir. EXCEPT_KEY: Kafka mesaj değeri key.fields içinde tanımlananlar hariç tüm tablo sütunlarını içerir.

• Kaynak tablo

  Seçenek	Açıklama	Tür	Gerekli	Varsayılan	Açıklamalar
  topic	Okunacak konu veya konular.	String	Hayır	–	Birden fazla konuya abone olmak için konu adlarını noktalı virgül (;) ile ayırın; örneğin, 'topic-1;topic-2'. Not Bu seçeneği veya topic-pattern seçeneğini belirtebilirsiniz ancak her ikisini aynı anda belirtemezsiniz.
  topic-pattern	Abone olunacak konularla eşleşen bir düzenli ifade. Tüketici, adları bu desenle eşleşen tüm konulara abone olur.	String	Hayır	–	Örnekler: user_event_.*: user_event_ önekiyle başlayan tüm konularla eşleşir. prod\.logs\..*: prod.logs. önekiyle başlayan konularla eşleşir (. karakteri kaçış karakteriyle belirtilmelidir). Not Bu seçeneği veya topic seçeneğini belirtebilirsiniz ancak her ikisini aynı anda belirtemezsiniz.
  properties.group.id	Kafka kaynak tüketici grubu kimliği.	String	Hayır	KafkaSource-{Source-Table-Name}	Bir tüketici grubu kimliğini ilk kez kullanıyorsanız başlangıç ofsetini tanımlamak amacıyla properties.auto.offset.reset değerini earliest veya latest olarak ayarlamalısınız.
  scan.startup.mode	Kafka tüketici başlangıç ofseti.	String	Hayır	group-offsets	Geçerli değerler: earliest-offset: En erken mevcut ofsetten okumaya başlar. latest-offset: En son ofsetten okumaya başlar. group-offsets: Belirtilen properties.group.id için commit edilmiş ofsetlerden okumaya başlar. timestamp: Belirtilen scan.startup.timestamp-millis değerinden okumaya başlar. specific-offsets: scan.startup.specific-offsets içinde belirtilen ofsetlerden okumaya başlar. Not Bu seçenek yalnızca durumu olmayan bir iş başladığında geçerlidir. Bir iş checkpoint'ten devam ediyorsa checkpoint durumunda saklanan ofsetlerden okur.
  scan.startup.specific-offsets	scan.startup.mode değeri specific-offsets olduğunda bölüm başına başlangıç ofseti.	String	Hayır	–	Örneğin, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	scan.startup.mode değeri timestamp olarak ayarlandığında milisaniye cinsinden başlangıç zaman damgası.	Long	Hayır	–	Birim milisaniyedir.
  scan.topic-partition-discovery.interval	Bölüm keşif aralığı.	Duration	Hayır	5 dakika	Bağlayıcı periyodik olarak yeni bölümleri keşfeder ve bunlardan okur. topic-pattern kullandığınızda bağlayıcı ayrıca desenle eşleşen yeni konuları da keşfeder. Bu özelliği devre dışı bırakmak için aralığı negatif olmayan bir değere ayarlayın. Not Ververica Runtime (VVR) 6.0.x'te dinamik bölüm keşfi varsayılan olarak devre dışıdır. VVR 8.0'dan itibaren bu özellik varsayılan olarak etkindir ve keşif aralığı 5 dakikadır.
  scan.header-filter	Kafka mesaj başlıklarına göre mesajları filtreler.	String	Hayır	–	Bir başlık anahtarı ve değeri iki nokta üst üste (:) ile ayrılır. Birden fazla başlık koşulu mantıksal operatörler (& ve |) kullanılarak bağlanır. NOT mantıksal operatörü (!) de desteklenir. Örneğin, depart:toy|depart:book&!env:test ifadesi başlık depart=toy veya depart=book içeriyor ve env=test içermiyorsa Kafka verilerini tutar. Not Bu seçenek yalnızca Ververica Runtime (VVR) 8.0.6 ve sonraki sürümlerde desteklenir. İfadelerde parantezler desteklenmez. Mantıksal işlemler soldan sağa değerlendirilir. Karşılaştırma için başlık değerleri UTF-8 dizelerine dönüştürülür.
  scan.check.duplicated.group.id	properties.group.id değerini zaten kullanan başka bir etkin tüketici olup olmadığını kontrol eder.	Boolean	Hayır	false	Geçerli değerler: true: İş başlamadan önce sistem yinelenen bir tüketici grubunu kontrol eder. Çakışmaları önlemek için böyle bir grup bulunursa iş başarısız olur. false: Çakışma kontrolü yapmadan işi başlatır. Not Bu seçenek yalnızca Ververica Runtime (VVR) 6.0.4 ve sonraki sürümlerde desteklenir.

• Hedef tablo

  Seçenek	Açıklama	Tür	Gerekli	Varsayılan	Açıklamalar
  topic	Hedef konu.	String	Evet	–	–
  sink.partitioner	Paralel hedef örneklerinden gelen kayıtları Kafka bölümlerine eşler.	String	Hayır	default	Geçerli değerler: default: Varsayılan Kafka bölümleyiciyi kullanır. fixed: Her paralel hedef örneği sabit bir Kafka bölümüne yazar. round-robin: Kayıtlar bölümlere round-robin yöntemiyle dağıtılır. Özel bölümleyici: Özel bir bölümleyici kullanmak için bir FlinkKafkaPartitioner alt sınıfının tam nitelikli sınıf adını sağlayın; örneğin, org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Hedef teslimat garantisi.	String	Hayır	at-least-once	Geçerli değerler: none: Garanti sağlamaz. Kayıtlar kaybolabilir veya yinelenebilir. at-least-once: Hiçbir kaydın kaybolmadığını garanti eder ancak kayıtlar yinelenebilir. exactly-once: Exactly-once semantiği sağlamak için Kafka transaction'larını kullanır ve kayıtların ne kaybolmasını ne de yinelenmesini sağlar. Not exactly-once semantiği kullanırken ayrıca sink.transactional-id-prefix belirtmeniz gerekir.
  sink.transactional-id-prefix	Transaction ID öneki. sink.delivery-guarantee değeri exactly-once olduğunda gereklidir.	String	Evet, sink.delivery-guarantee değeri exactly-once ise	–	Yalnızca sink.delivery-guarantee değeri exactly-once olarak ayarlandığında gereklidir.
  sink.parallelism	Hedef operatör paralelliği.	Integer	Hayır	–	Varsayılan olarak framework paralelliği upstream operatörlere göre belirler.

Güvenlik ve kimlik doğrulama

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* SSL configuration */
  /* Path to the truststore for the server's CA certificate. */
  /* Files uploaded using Artifacts are stored in the /flink/usrlib/ directory. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* If client authentication is required, you must also configure the path to the keystore (private key). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* The algorithm used to verify the server hostname. An empty string disables hostname verification. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* SASL configuration */
  /* Set the SASL mechanism to SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Configure JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

Örnekte bahsedilen CA sertifikasını ve özel anahtarı yüklemek için Realtime Compute for Apache Flink konsolunun Artifacts özelliğini kullanabilirsiniz. Yüklenen dosyalar /flink/usrlib dizininde saklanır. my-truststore.jks adlı bir CA sertifika dosyası kullanmak için WITH yan tümcesinde 'properties.ssl.truststore.location' özelliğini aşağıdaki iki yoldan biriyle ayarlayabilirsiniz:

• 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' ayarlayın. Bu yöntem çalışma zamanında Object Storage Service (OSS) üzerinden dosyaların dinamik olarak indirilmesini önler ancak Hata Ayıklama Modunu desteklemez.

• Realtime Compute altyapı sürümü VVR 11.5 veya sonraki bir sürümse properties.ssl.truststore.location ve properties.ssl.keystore.location değerlerini mutlak bir OSS yoluna ayarlayabilirsiniz. Dosya yolu formatı oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Namespace name>/<file name> şeklindedir. Bu yöntem Flink çalışma zamanı sırasında OSS dosyalarını dinamik olarak indirir ve Hata Ayıklama Modunu destekler.

Kaynak tablo başlangıç ofseti

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume from the committed offsets of the consumer group "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- If "my-group" is used for the first time, consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consumption starts from the latest offset.
  -- Consume from the specified timestamp in milliseconds: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from specific offsets.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Kaynak ofsetlerini commit etme

Kafka kaynak tablosu tüketici ofsetini Kafka kümesine yalnızca başarılı bir checkpoint'ten sonra commit eder; bu nedenle uzun bir checkpoint aralığı commit edilen ofsetin geride kalmasına neden olur. Kaynak tablo gerçek okuma ilerlemesini checkpoint durumunda saklar ve sistem bunu hata kurtarma için kullanır. Commit edilen ofsetler yalnızca ilerleme izleme amaçlıdır ve kurtarma için kullanılmaz; bu nedenle commit hataları veri doğruluğunu etkilemez.

Özel hedef bölümleyici

Kafka'nın yerleşik bölümleme stratejisi gereksinimlerinizi karşılamıyorsa FlinkKafkaPartitioner sınıfını genişleterek özel bir bölümleyici uygulayabilirsiniz. Geliştirme tamamlandıktan sonra kodunuzu bir JAR paketine derleyin ve Realtime Compute konsolundaki Artifacts özelliğini kullanarak yükleyin. JAR paketi yüklendikten ve referans verildikten sonra WITH yan tümcesinde sink.partitioner parametresini bölümleyicinizin tam nitelikli sınıf adına ayarlayın; örneğin, org.mycompany.MyPartitioner.

Kafka, Upsert Kafka ve Kafka JSON katalog

Kafka yalnızca ekleme yapılabilen bir olay akışı platformudur ve veri güncellemelerini veya silmelerini desteklemez. Akış SQL'inde standart bir Kafka hedef tablosu upstream Change Data Capture (CDC) verilerini veya aggregate ve join gibi operatörlerin geri çekme mantığını işleyemez. Değişiklik veya geri çekme içeren verileri yazmanız gerekiyorsa bir Upsert Kafka hedef tablosu kullanın.

Bir veya daha fazla upstream veritabanı tablosundan Kafka'ya Change Data Capture (CDC) verilerinin toplu eşzamanlamasını basitleştirmek için Kafka JSON katalog kullanabilirsiniz. Kafka'da saklanan veriler JSON formatındaysa Kafka JSON katalog şema ve WITH parametrelerini tanımlama adımını atlamanızı sağlar. Ayrıntılar için bkz. Manage Kafka JSON catalogs.

Örnekler

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- Optional. Flattens all nested columns.
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--Use COALESCE to handle null values.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

Exactly-once semantiği

• Tüketici izolasyon düzeyini yapılandırın

  Kafka verilerini tüketen tüm uygulamalar isolation.level özelliğini ayarlamalıdır:

  • read_committed: Yalnızca commit edilmiş verileri okur.

  • read_uncommitted (Varsayılan): Commit edilmemiş verileri okuyabilir.

  EXACTLY_ONCE read_committed değerine bağlıdır. Aksi takdirde tüketiciler commit edilmemiş verileri görebilir ve tutarlılık bozulur.

• Transaction zaman aşımı ve veri kaybı

  Bir checkpoint'ten kurtulurken Realtime Compute for Apache Flink yalnızca o checkpoint başlamadan önce commit edilmiş transaction'ları dikkate alır. Bir iş arızası ile yeniden başlatılması arasındaki süre Kafka transaction zaman aşımını aşarsa Kafka açık transaction'ı otomatik olarak iptal eder ve bu durum veri kaybına yol açabilir.

  • Bir Kafka broker için varsayılan transaction.max.timeout.ms değeri 15 dakikadır.

  • Flink Kafka Sink varsayılan olarak transaction.timeout.ms parametresini 1 saat olarak ayarlar.

  • Broker üzerindeki transaction.max.timeout.ms değerini Flink'teki ayara eşit veya ondan büyük olacak şekilde artırmalısınız.

• Üretici havuzu ve eşzamanlı checkpoint'ler

  EXACTLY_ONCE modu sabit boyutlu bir Kafka üretici havuzu kullanır. Her checkpoint bu havuzdan bir üretici kullanır. Eşzamanlı checkpoint sayısı havuz boyutunu aşarsa iş başarısız olur.

  Üretici havuzu boyutunu maksimum eşzamanlı checkpoint sayısına göre yapılandırın.

• Paralellik azaltma kısıtlamaları

  Bir iş ilk checkpoint tamamlanmadan önce başarısız olursa orijinal üretici havuzu bilgileri yeniden başlatmada kaybolur. Bu nedenle ilk checkpoint tamamlanana kadar işin paralelliğini azaltmayın. Azaltma gerekliyse yeni paralellik FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR değerinden düşük olmamalıdır.

• Transaction'lar okumaları engeller

  read_committed modunda commit edilmemiş veya iptal edilmemiş herhangi bir transaction konudaki tüm okuma işlemlerini engeller.

  Örneğin:

  • Transaction 1 veri yazar.

  • Transaction 2 daha fazla veri yazar ve commit edilir.

  • Transaction 1 açık kaldığı sürece commit edilmiş Transaction 2'den gelen veriler tüketiciler tarafından görünmez.

  Bunun aşağıdaki sonuçları vardır:

  • Normal çalışma sırasında veri görünürlük gecikmesi yaklaşık olarak checkpoint aralığına eşittir.

  • Bir iş başarısız olursa yazdığı konu iş yeniden başlatılana veya transaction zaman aşımına uğrayana kadar tüketiciler tarafından engellenir. Aşırı durumlarda transaction zaman aşımı süreci de okuma işlemlerini etkileyebilir.

SSS

• Realtime Compute for Apache Flink kullanarak Kafka'dan JSON verilerini nasıl alırım?

• ApsaraMQ for Kafka Kaynak Tablosu için Event Time tabanlı Pencere neden çıktı üretmez?

• Güvenli bir Kafka Kümesine nasıl bağlanırım?

• Alan adı çakışmalarını nasıl çözerim?

• Workspaces and Namespaces