Realtime Compute for Apache Flink:Lindorm connector

Önkoşullar

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

• Bir Lindorm geniş tablo motoru ve bir Lindorm tablosu. Daha fazla bilgi için bkz. Create an instance.

• Lindorm kümesi ile Flink çalışma alanı arasında ağ bağlantısı (örneğin, her ikisinin de aynı sanal özel bulut (VPC) içinde bulunması).

Lindorm HBase tabloları desteklenmez. Yalnızca LindormTable desteklenir.

Hızlı başlangıç

Aşağıdaki örnek 10 satır veri oluşturur, bir Lindorm boyut tablosunda eşleşen satırları arar ve sonucu bir Lindorm alıcı tablosuna yazar.

-- Source: generate 10 rows with sequential IDs 0-9
CREATE TEMPORARY TABLE example_source (
  id INT,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '10',
  'fields.id.kind' = 'sequence',
  'fields.id.start' = '0',
  'fields.id.end' = '9'
);

-- Dimension table: look up user details from Lindorm
CREATE TEMPORARY TABLE lindorm_hbase_dim (
  `id`    INT,
  `name`  VARCHAR,
  `birth` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'   = 'lindorm',
  'tablename'   = '${lindorm_dim_table}',
  'seedserver'  = '${lindorm_seed_server}',
  'namespace'   = 'default',
  'username'    = '${lindorm_username}',
  'password'    = '${lindorm_password}'
);

-- Sink table: write enriched records to Lindorm
CREATE TEMPORARY TABLE lindorm_hbase_sink (
  `id`    INT,
  `name`  VARCHAR,
  `birth` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'   = 'lindorm',
  'tablename'   = '${lindorm_sink_table}',
  'seedserver'  = '${lindorm_seed_server}',
  'namespace'   = 'default',
  'username'    = '${lindorm_username}',
  'password'    = '${lindorm_password}'
);

-- Temporal join: enrich source data with the dimension table and write to sink
INSERT INTO lindorm_hbase_sink
SELECT
  s.id,
  d.name,
  d.birth
FROM example_source AS s
JOIN lindorm_hbase_dim AS d FOR SYSTEM_TIME AS OF s.proc_time
  ON s.id = d.id;

Çalıştırmadan önce yer tutucuları değiştirin:

Söz dizimi

CREATE TABLE white_list (
  id     VARCHAR,
  name   VARCHAR,
  age    INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'    = 'lindorm',
  'seedserver'   = '<host:port>',
  'namespace'    = '<yourNamespace>',
  'username'     = '<yourUsername>',
  'password'     = '<yourPassword>',
  'tableName'    = '<yourTableName>',
  'columnFamily' = '<yourColumnFamily>'
);

Connector seçenekleri

Arama önbelleği

Varsayılan olarak connector her aramayı doğrudan Lindorm'dan getirir (cache = None). Yüksek aktarım hızına sahip join işlemlerinde Lindorm okuma yükünü azaltmak için LRU önbelleğe almayı etkinleştirin.

cache = LRU ayarıyla connector son erişilen satırları bellekte saklar. Join anahtarı önbellekteki bir satırla eşleştiğinde connector Lindorm'u sorgulamadan önbelleğe alınan değeri döndürür. Bir önbellek girdisi şu durumlarda kaldırılır:

• Önbellek cacheSize satıra ulaştığında (en eski satırlar önce kaldırılır).

• Girdi önbellekte cacheTTLMs milisaniyeden daha uzun süre kaldığında.

Bir ödünleşim söz konusudur: Daha uzun bir TTL veya daha büyük bir önbellek Lindorm okuma trafiğini azaltır ancak eskimiş veri sunma riskini artırır. Her iki değeri de aktarım hızı gereksinimlerinize ve temel verilerin ne kadar hızlı değiştiğine göre ayarlayın.

Lindorm connector bire çok arama joinlerini destekler. Boyut tablosundaki satırlar birden fazla yukarı akış olayıyla eşleşebiliyorsa önbelleğe alma stratejilerine ve aktarım hızına özellikle dikkat edin.

Tekil yazmalar

Bir Lindorm tablosunda birincil anahtar bulunduğunda tüm yazmalar upsert semantiği kullanır: Gelen her kayıt ya yeni bir satır ekler ya da mevcut eşleşen satırı günceller. Bu durum yazmaları tekil (idempotent) hale getirir.

Tekil yazmalar hata toleransı için önemlidir. Bir Flink işi bir kontrol noktasından yeniden başlatıldığında son başarılı kontrol noktasından itibaren mesajları yeniden oynatır. Lindorm upsert işlemleri tekil olduğu için yeniden oynatılan kayıtlar orijinal yazmalarla aynı sonucu üretir; yinelenen satırlar veya kural ihlalleri oluşmaz.

Tekil yazmalardan yararlanmak için DDL'nizde bir birincil anahtar tanımlayın.

Dinamik tablo

Şemanız çalışma zamanında değişiyorsa dinamik tablo özelliğini kullanın; sütunlar DDL zamanında sabitlenmek yerine verideki değerlere göre dinamik olarak oluşturulur. Tipik bir kullanım durumu, saatlerin sütun adı günlerin ise birincil anahtar olduğu günlük saatlik ölçümlerin izlenmesidir:

Dinamik tablolar için DDL kuralları:

• İlk N sütun birincil anahtarı oluşturur.

• Son iki sütun VARCHAR türünde olmalıdır.

• Sondan bir önceki sütun (c1) dinamik sütun adını tutar.

• Son sütun (c2) o sütun için değeri tutar.

• c1 ve c2 dışındaki birincil anahtar olmayan sütunlara izin verilmez.

CREATE TABLE lindorm_dynamic_output (
  pk1 VARCHAR,
  pk2 VARCHAR,
  pk3 VARCHAR,
  c1  VARCHAR,  -- column name written to Lindorm
  c2  VARCHAR,  -- column value written to Lindorm
  PRIMARY KEY (pk1, pk2, pk3) NOT ENFORCED
) WITH (
  'connector'         = 'lindorm',
  'seedserver'        = '<host:port>',
  'namespace'         = '<yourNamespace>',
  'username'          = '<yourUsername>',
  'password'          = '<yourPassword>',
  'tableName'         = '<yourTableName>',
  'columnFamily'      = '<yourColumnFamily>',
  'dynamicColumnSink' = 'true'
);

Her kayıt yazıldığında connector <pk1, pk2, pk3> ile tanımlanan Lindorm satırına bir sütun ekler veya bu sütunu günceller. O satırdaki diğer sütunlar değişmeden kalır.

Veri türü eşlemeleri

Tüm Lindorm verileri ikili biçimde depolanır. Aşağıdaki tablo connector'un Flink SQL türleri ile Lindorm ikili temsilleri arasında nasıl dönüşüm yaptığını gösterir.

Bytes yöntemleri com.alibaba.lindorm.client.core.utils.Bytes sınıfındadır. StringData yöntemleri (CHAR/VARCHAR satırları) org.apache.flink.table.data.StringData sınıfındadır.

Ölçümler

Aşağıdaki alıcı ölçümleri kullanılabilir. Ayrıntılar için bkz. Metrics.

SSS

Lindorm connection errors and solutions