Realtime Compute for Apache Flink:Upsert Kafka

Ön koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

• Bir Kafka kümesi. Bir DataFlow Kafka kümesi oluşturun veya ApsaraMQ for Kafka'da kaynaklar oluşturun

• Flink kümeniz ile Kafka kümeniz arasında ağ bağlantısı. EMR üzerindeki Kafka için bir VPC ve güvenlik grubu yapılandırın. ApsaraMQ for Kafka için bir izin listesi yapılandırın

Nasıl çalışır?

Kaynak tarafından üretilen upsert değişiklik günlüğü akışı yalnızca +I (insert), +U (update after) ve -D (delete) girişlerini içerir. Geri çekme (retract) akışının aksine, hiçbir zaman -U (update before) mesajları içermez.

Kaynak tablo olarak, bağlayıcı Kafka kayıtlarını okur ve bunları bir değişiklik günlüğü akışına dönüştürür:

• Anahtarı zaten mevcut olan bir kayıt UPDATE (+U) olarak yorumlanır

• Anahtarı mevcut olmayan bir kayıt INSERT (+I) olarak yorumlanır

• Null değerli bir kayıt DELETE (-D) olarak yorumlanır

Hedef tablo veya veri alımı hedefi olarak, bağlayıcı upstream değişiklik günlüğü akışını tüketir:

• INSERT ve UPDATE_AFTER kayıtları normal Kafka mesajları olarak yazılır

• DELETE kayıtları tombstone mesajları (ilgili anahtar için null değer) olarak yazılır

Flink, hedef çıkışını birincil anahtara göre bölmeye ayırır; böylece aynı anahtara ait tüm mesajlar doğru sırada aynı bölüme ulaşır.

Örnek

Aşağıdaki uçtan uca örnek, bir Kafka kaynak tablosundan web sitesi sayfa görüntüleme olaylarını okur, bölge başına sayfa görüntülemelerini (PV) ve benzersiz ziyaretçileri (UV) toplar ve sonuçları bir Upsert Kafka hedefine yazar.

Kaynak tablo — ham sayfa görüntüleme olaylarını okur:

CREATE TABLE pageviews (
  user_id    BIGINT,
  page_id    BIGINT,
  viewtime   TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector'                  = 'kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'format'                     = 'json'
);

Hedef tablo — upsert semantiğiyle toplanmış sonuçları depolar:

CREATE TABLE pageviews_per_region (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '<host:port>',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Her Upsert Kafka tablosu bir birincil anahtar tanımlamalıdır. Birincil anahtar sütunları, Flink'in çıkışı nasıl böldüğünü ve bağlayıcının upsert işlemlerini nasıl tanımladığını belirler.

Sorgu — toplama yapar ve sonuçları yazar:

INSERT INTO pageviews_per_region
SELECT
  user_region,
  COUNT(*),
  COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;

SQL

CREATE TABLE upsert_kafka_sink (
  user_region STRING,
  pv          BIGINT,
  uv          BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector'                  = 'upsert-kafka',
  'topic'                      = '<yourTopicName>',
  'properties.bootstrap.servers' = '...',
  'key.format'                 = 'avro',
  'value.format'               = 'avro'
);

Veri alımı

Upsert Kafka bağlayıcısı, bir YAML veri alımı işinde hedef olarak kullanılabilir. Veriler JSON biçiminde yazılır ve birincil anahtar alanları mesaj gövdesine dahil edilir.

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # ApsaraMQ for Kafka
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: upsert-kafka
  name: Upsert Kafka Sink
  properties.bootstrap.servers: ${upsert.kafka.bootstraps.server}
  aliyun.kafka.accessKeyId: ${upsert.kafka.aliyun.ak}
  aliyun.kafka.accessKeySecret: ${upsert.kafka.aliyun.sk}
  aliyun.kafka.instanceId: ${upsert.kafka.aliyun.instanceid}
  aliyun.kafka.endpoint: ${upsert.kafka.aliyun.endpoint}
  aliyun.kafka.regionId: ${upsert.kafka.aliyun.regionid}

route:
  - source-table: ${mysql.source.table}
    sink-table: ${upsert.kafka.topic}

Teslimat semantiği

Varsayılan olarak, Upsert Kafka hedefi at-least-once garantisiyle yazar. Flink, aynı anahtara sahip çoğaltılmış kayıtları konuya yazabilir. Bağlayıcı upsert modunda çalıştığından, konu bir kaynak olarak tekrar okunduğunda herhangi bir anahtar için son kayıt geçerli olur. Bu nedenle çoğaltılmış yazmalar idempotenttir — nihai durumu bozmazlar.

Exactly-once semantiğini etkinleştirmek için Kafka kümesinin Kafka işlemlerini (Apache Kafka 0.11 veya sonraki sürümler) desteklemesi ve işlem özelliğinin etkinleştirilmiş olması gerekir. Veri alımı YAML yapılandırmasında sink.delivery-guarantee değerini exactly-once olarak ayarlayın.

Sınırlamalar

• Ververica Runtime (VVR) 2.0.0 veya sonraki bir sürümde çalışan Flink gerektirir.

• Hem okuma hem de yazma işlemleri için Apache Kafka 0.10 veya sonraki sürümleri destekler.

• Yalnızca Apache Kafka 2.8 için belgelenen istemci parametrelerini destekler. Bkz. tüketici ve üretici yapılandırma referansları.

• Upsert Kafka kaynak tabloları her zaman earliest-offset konumundan başlar — bu sabittir ve yapılandırılamaz. Bağlayıcının eksiksiz bir değişiklik günlüğünü yeniden oluşturabilmesi için tüm geçmiş değişiklik verilerini okuması gerekir. Diğer herhangi bir başlatma modunun (latest-offset gibi) veya zaman damgasına göre başlatmanın kullanılması, eksik bir değişiklik günlüğüyle sonuçlanır ve downstream hesaplamalarda veri doğruluğu sorunlarına neden olur.

• Exactly-once hedef semantiği için, hedef Kafka kümesinin işlemler etkinleştirilmiş Apache Kafka 0.11 veya sonraki bir sürümü çalıştırması gerekir.

İzleme ölçümleri

Kaynak tablo ölçümleri

Hedef tablo ölçümleri

Sonraki adımlar

• Flink CDC kullanarak tüm bir MySQL veritabanını Kafka'ya senkronize etme

• Flink bağımlılık çakışmalarını çözme