StarRocks SQL connector - Realtime Compute for Apache Flink

Arka Plan

StarRocks, tüm senaryolarda son derece hızlı performans ve birleştirilmiş bir analitik deneyimi sunan yeni nesil Massively Parallel Processing (MPP) veri ambarıdır. StarRocks aşağıdaki avantajları sunar:

StarRocks MySQL protokolü ile uyumludur; bu sayede MySQL istemcilerini ve yaygın Business Intelligence (BI) araçlarını kullanarak verilere bağlanabilir ve verileri analiz edebilirsiniz.
StarRocks dağıtılmış bir mimari kullanır:
- Veri tablolarını yatay olarak böler ve birden çok çoğaltma ile saklar.
- Küme esnek şekilde ölçeklendirilebilir ve 10 petabayta (PB) kadar veriyi analiz edebilir.
- Paralel hesaplamaları hızlandırmak için bir MPP çerçevesi kullanır.
- Hata toleransı sağlamak için birden çok çoğaltmayı destekler.

Flink bağlayıcısı verileri önbelleğe alır ve toplu işler halinde sink tablolara yazmak için Stream Load'u kullanır. Kaynak tablolardan veri okurken verileri toplu olarak getirir. Aşağıdaki tabloda StarRocks bağlayıcısının yetenekleri listelenmiştir.

Kategori	Açıklama
Desteklenen türler	kaynak tablolar, boyut tabloları, sink tablolar ve veri alımı hedefleri
Çalıştırma modu	akış modu ve toplu iş modu
Veri biçimi	CSV
Bağlayıcıya özel ölçümler	Yok
API türleri	veri alımı için DataStream, SQL ve YAML
Sink tablolarda güncellemelerin/silmelerin desteklenmesi	Evet

Önkoşullar

EMR üzerinde dağıtılmış bir StarRocks kümesine veya ECS üzerinde kendi yönettiğiniz bir StarRocks kümesine sahip olmanız gerekir.

Sınırlamalar

Boyut tablolarıyla join işlemlerini yalnızca Ververica Runtime (VVR) 11,1 veya sonraki sürümleri destekler.
Ağ erişim kısıtlamalarını önlemek için aşağıdaki StarRocks küme bağlantı noktalarını bir güvenlik grubuna veya güvenlik duvarı izin listesine ekleyin: 9030, 8030, 8040, 9060, 8060, 9020.
Hedef StarRocks tablosunda gizli oluşturulmuş sütunlar bulunduğunda, Flink işinde yalnızca yazılacak gerçek fiziksel sütunları bildirin ve oluşturulmuş sütun alanlarını hariç tutun. StarRocks ifade bölümlemesi tarafından otomatik olarak oluşturulan gizli oluşturulmuş sütunlar (örneğin __generated_partition_column_0) harici yazmaları kabul etmez. Bağlayıcı varsayılan olarak tam şemaya göre yazma istekleri oluşturur ve bu durum işin başarısız olmasına neden olur. StarRocks'ta oluşturulmuş sütunlar hakkında daha fazla bilgi için bkz. Generated columns.

Özellikler

E-MapReduce üzerindeki StarRocks, CREATE TABLE AS SELECT (CTAS) ve CREATE DATABASE AS SELECT (CDAS) deyimlerini destekler. CTAS tek bir tablonun şemasını ve verilerini senkronize ederken CDAS bir veritabanının tamamını veya aynı veritabanındaki birden çok tabloyu senkronize eder. Daha fazla bilgi için bkz. Realtime Compute for Apache Flink'te CTAS ve CDAS deyimlerini kullanarak MySQL veritabanından StarRocks'a veri senkronize etme.

Sözdizimi

CREATE TABLE USER_RESULT(
 name VARCHAR,
 score BIGINT
 ) WITH (
 'connector' = 'starrocks',
 'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx',
 'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port',
 'database-name' = 'xxx',
 'table-name' = 'xxx',
 'username' = 'xxx',
 'password' = 'xxx'
 );

Parametreler

Tür	Parametre	Açıklama	Tür	Gerekli	Varsayılan	Açıklamalar
Genel	connector	Kullanılacak bağlayıcıyı belirtir.	Dize	Evet	—	Değer `starrocks` olmalıdır.
	jdbc-url	Java Database Connectivity (JDBC) URL'si.	Dize	Evet	—	FE'nin IP adresini ve JDBC bağlantı noktasını `jdbc:mysql://ip:port` biçiminde belirtin.
	database-name	StarRocks veritabanının adı.	Dize	Evet	—	—
	table-name	StarRocks tablosunun adı.	Dize	Evet	—	—
	username	StarRocks'a bağlanmak için kullanılan kullanıcı adı.	Dize	Evet	—	—
	password	StarRocks'a bağlanmak için kullanılan parola.	Dize	Evet	—	—
	starrocks.create.table.properties	Otomatik tablo oluşturma için özellikleri belirtir.	Dize	Hayır	—	Motor ve çoğaltma sayısı gibi ilk tablo özelliklerini belirtir. Örneğin: 'starrocks.create.table.properties' = 'buckets 8' veya 'starrocks.create.table.properties' = 'replication_num=1'.
Kaynağa özel	scan-url	Veri tarama URL'si.	Dize	Hayır	—	FE'nin IP adresini ve HTTP bağlantı noktasını belirtir. Biçim: `fe_ip:http_port;fe_ip:http_port`. Not Birden çok IP adresi ve bağlantı noktası belirtmek için bunları virgülle (,) ayırın.
	scan.connect.timeout-ms	`flink-connector-starrocks` bağlayıcısının StarRocks'a bağlanması için zaman aşımı süresi. Bağlantı bu zaman aşımı içinde kurulamazsa bağlayıcı bir hata bildirir.	Dize	Hayır	1000	Birim: milisaniye.
	scan.params.keep-alive-min	Sorgu görevi için keep-alive süresi.	Dize	Hayır	10	—
	scan.params.query-timeout-s	Sorgu görevi için zaman aşımı süresi. Bu süre içinde sonuç döndürülmezse sistem sorgu görevini durdurur.	Dize	Hayır	600	Birim: saniye.
	scan.params.mem-limit-byte	Bir BE düğümündeki tek bir sorgu için bellek sınırı.	Dize	Hayır	1073741824 (1 GB)	Birim: bayt.
	scan.max-retries	Başarısız bir sorgu için maksimum yeniden deneme sayısı. Bu sınır aşıldığında bağlayıcı bir hata bildirir.	Dize	Hayır	1	—
Sink'e özel	load-url	Veri içe aktarma URL'si.	Dize	Evet	—	FE'lerin IP adreslerini ve HTTP bağlantı noktalarını `fe_ip:http_port;fe_ip:http_port` biçiminde belirtin. Not Birden çok IP adresi ve bağlantı noktası belirtmek için bunları noktalı virgülle (;) ayırın.
	sink.semantic	Yazmalar için teslim semantiği.	Dize	Hayır	at-least-once	Geçerli değerler: `at-least-once` (varsayılan): Verilerin en az bir kez teslim edilmesini garanti eder. `exactly-once`: Verilerin tam olarak bir kez teslim edilmesini garanti eder.
	sink.buffer-flush.max-bytes	Temizlemeden önce arabelleğe alınacak maksimum veri miktarı.	Dize	Hayır	94371840 (90 MB)	Geçerli aralık: 64 MB ila 10 GB.
	sink.buffer-flush.max-rows	Temizlemeden önce arabelleğe alınacak maksimum satır sayısı.	Dize	Hayır	500000	Geçerli aralık: 1.000 ila 5.000.000.
	sink.buffer-flush.interval-ms	Arabellek temizleme aralığı.	Dize	Hayır	300000	Geçerli aralık: 1.000 ms ila 3.600.000 ms.
	sink.max-retries	Başarısız yazmalar için maksimum yeniden deneme sayısı.	Dize	Hayır	3	Geçerli aralık: 0 ila 1000.
	sink.connect.timeout-ms	StarRocks'a bağlanmak için zaman aşımı süresi.	Dize	Hayır	1000	Geçerli aralık: 100 ila 60.000. Birim: milisaniye.
	sink.properties.*	Sink için ek Stream Load özellikleri.	Dize	Hayır	—	Bu parametreler Stream Load davranışını kontrol eder. Örneğin, `sink.properties.format`, CSV gibi içe aktarılan verilerin biçimini belirtir. Daha fazla parametre için bkz. Stream Load.
Boyuta özel	lookup.cache.enabled	Boyut tablosu için önbelleğe almanın etkinleştirilip etkinleştirilmeyeceğini belirtir.	Boolean	Hayır	true	Geçerli değerler: `true`: Önbelleğe almayı etkinleştirir. Tablo verileri ilk kez okunduktan sonra sistem bunları bellekte önbelleğe alır. Sonraki istekler, G/Ç yükünü azaltmak için geçerlilik süresi içinde önbelleğe alınan verileri kullanır. `false`: Önbelleğe almayı devre dışı bırakır. Her sorgu doğrudan veri kaynağına erişir. Önemli Bu özellik Realtime Compute for Apache Flink motoru VVR 11,1 ve sonraki sürümlerini gerektirir. Aşağıdaki senaryolarda bu özelliği devre dışı bırakmanızı öneririz: Boyut tablosu verileri sık sık güncellenir ve gerçek zamanlı veri gerekir. Tablo büyük miktarda veri içerir ve bellek taşması riski oluşur.

Veri türü eşlemesi

StarRocks veri türü	Flink veri türü
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Not Realtime Compute for Apache Flink motoru VVR 8.0.10 ve sonraki sürümlerini gerektirir.	DECIMAL(20,0)
LARGEINT	DECIMAL(20,0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DATE	DATE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
DECIMALV2	DECIMAL
DECIMAL32	DECIMAL
DECIMAL64	DECIMAL
DECIMAL128	DECIMAL
CHAR(m) Not VVR 8.0.10, MySQL ve StarRocks arasındaki kodlama farklarını karşılamak için CHAR uzunluğunu otomatik olarak üç katına çıkarır (m=n3, n<=85). VVR 8.0.11 ve sonraki sürümler, MySQL ve StarRocks arasındaki kodlama farklarını karşılamak için CHAR uzunluğunu otomatik olarak dört katına çıkarır (m=n4, n<=63). StarRocks CHAR türü için maksimum uzunluk 255'tir. Bu nedenle, Flink bir CHAR türünü yalnızca otomatik olarak genişletilen uzunluğu 255'i aşmadığında StarRocks CHAR türüne eşler.	CHAR(n)
VARCHAR(m) Not VVR 8.0.10, MySQL ve StarRocks arasındaki kodlama farklarını karşılamak için VARCHAR uzunluğunu otomatik olarak üç katına çıkarır (m=n3, n>85). VVR 8.0.11 ve sonraki sürümler, MySQL ve StarRocks arasındaki kodlama farklarını karşılamak için VARCHAR uzunluğunu otomatik olarak dört katına çıkarır (m=n4, n>63). StarRocks CHAR türü için maksimum uzunluk 255'tir. Bu nedenle, bir Flink CHAR türünün otomatik olarak genişletilen uzunluğu 255'i aşarsa Flink türü StarRocks VARCHAR türüne eşler.	CHAR(n)
VARCHAR	STRING
VARBINARY Not Realtime Compute for Apache Flink motoru VVR 8.0.10 ve sonraki sürümlerini gerektirir.	VARBINARY

Örnekler

Aşağıdaki örneklerdeki kaynak tablo, sink tablo ve boyut tablosu birlikte kullanılabilir. Kaynak tablo örneğinde bildirilen runoob_tbl_source tablosu, sink tablo ve boyut tablosu örnekleri için de veri kaynağı olarak hizmet eder. Bu örnekleri kullanmadan önce StarRocks'ta eşleşen alan türlerine sahip tablolar hazırlayın ve uç noktaları, veritabanı ve tablo adlarını ile gizli değişkenleri değiştirin.

Kaynak tablo

CREATE TEMPORARY TABLE runoob_tbl_source (
    runoob_id BIGINT NOT NULL,
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    proc_time AS PROCTIME() --Processing time column, used by the temporal join in the dimension table example
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<source_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'scan.params.query-timeout-s' = '600' --Timeout for a single query, in seconds
);

Not

Kaynak tablo toplu iş modunda okunur. İş başladıktan sonra tablonun mevcut verilerini tarar ve ardından çıktı üretmeyi durdurur. Artımlı değişiklikleri sürekli olarak üretmez.

Sink tablo

Not

StarRocks, birincil anahtar sütununun NULLABLE olmasına izin verir. Ancak Flink, boş bırakılabilir bir sütun içeren birincil anahtarı desteklemez. Flink'in veri tutarlılığı modeli, birincil anahtarın benzersiz ve boş bırakılamaz olmasını gerektirir. Aksi takdirde Flink Invalid primary key. Column 'xxx' is nullable hatasını verir. Daha fazla bilgi için bkz. "Invalid primary key. Column 'xxx' is nullable." hatası.

VVR 11+

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.version' = 'V2', --Transactional Stream Load. Requires StarRocks 2.4 or later.
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000' --Batch flush interval, in milliseconds
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

VVR 8+

VVR 8.x'te sink.version varsayılan değeri V1'dir ve V1, V2 ve AUTO desteklenir. Yalnızca VVR 11.x'te kayıtlı olan sink.ignore.update-before gibi parametreler desteklenmez:

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000', --Batch flush interval, in milliseconds
    'sink.max-retries' = '3' --Number of retries after a Stream Load failure
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

Boyut tablosu

Boyut tablosu join işlemleri yalnızca VVR 11,1 ve sonraki sürümlerde desteklenir. Boyut tablosu, join anahtarı olarak bir birincil anahtar bildirmelidir ve zamansal join işlemi işleme zamanını kullanır. Aşağıdaki örnek, proc_time işleme zamanı sütununu içeren kaynak tablo örneğindeki runoob_tbl_source tablosunu temel alır:

CREATE TEMPORARY TABLE sr_dim (
    runoob_id BIGINT NOT NULL,
    runoob_author STRING,
    PRIMARY KEY (runoob_id) NOT ENFORCED --The dimension table must declare a primary key as the join key
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<dim_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'lookup.cache.ttl-ms' = '5000', --TTL of the dimension table cache, in milliseconds
    'lookup.cache.enabled' = 'true' --Default: true. Set to false to query the table directly without the cache.
);

SELECT o.runoob_id, o.runoob_title, d.runoob_author
FROM runoob_tbl_source AS o
JOIN sr_dim FOR SYSTEM_TIME AS OF o.proc_time AS d --Temporal join that uses the processing time column of the source table
ON o.runoob_id = d.runoob_id;