Realtime Compute for Apache Flink:StarRocks

Fitur

StarRocks pada E-MapReduce mendukung pernyataan CREATE TABLE AS SELECT (CTAS) dan CREATE DATABASE AS SELECT (CDAS). CTAS menyinkronkan skema dan data dari satu tabel, sedangkan CDAS menyinkronkan seluruh database atau beberapa tabel dalam database yang sama. Untuk informasi lebih lanjut, lihat Gunakan pernyataan CTAS dan CDAS di Realtime Compute for Apache Flink untuk menyinkronkan data dari database MySQL ke StarRocks.

Sintaks

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'
 );

Parameter

Pemetaan tipe data

Contoh kode

CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_source` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
) WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'scan-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxxx'
);
CREATE TEMPORARY TABLE IF NOT EXISTS `runoob_tbl_sink` (
  `runoob_id` BIGINT NOT NULL,
  `runoob_title` STRING NOT NULL,
  `runoob_author` STRING NOT NULL,
  `submission_date` DATE NULL
  PRIMARY KEY(`runoob_id`)
  NOT ENFORCED
) WITH (
  'jdbc-url' = 'jdbc:mysql://ip:9030',
  'connector' = 'starrocks',
  'load-url' = 'ip:18030',
  'database-name' = 'db_name',
  'table-name' = 'table_name',
  'password' = 'xxxxxxx',
  'username' = 'xxxx',
  'sink.buffer-flush.interval-ms' = '5000'
);

INSERT INTO runoob_tbl_sink SELECT * FROM runoob_tbl_source;

Fitur

• Pembuatan database dan tabel otomatis.

  Jika database atau tabel hulu tidak ada di instans StarRocks hilir, konektor akan membuatnya secara otomatis. Anda dapat menggunakan parameter table.create.properties.* untuk mengonfigurasi opsi pembuatan tabel otomatis.

• Sinkronisasi perubahan skema.

  Konektor StarRocks secara otomatis menerapkan event CreateTableEvent, AddColumnEvent, dan DropColumnEvent ke database hilir.

• VVR 11.1 dan yang lebih baru mendukung perubahan tipe kolom yang kompatibel. Untuk informasi lebih lanjut, lihat ALTER TABLE | StarRocks.

Sintaks

source:
  ...

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://127.0.0.1:9030
  load-url: 127.0.0.1:8030
  username: root
  password: pass
  sink.buffer-flush.interval-ms: 5000   # Atur interval flush data.

Konfigurasi

Gunakan katalog bawaan yang telah ada

VVR 11.5 dan yang lebih baru memungkinkan Anda mereferensikan katalog StarRocks bawaan yang dibuat di halaman Data Management secara langsung dalam pekerjaan ingesti data Flink CDC. Hal ini menyederhanakan konfigurasi dengan mengurangi jumlah properti yang perlu Anda atur secara manual.

sink:
  type: starrocks
  using.built-in-catalog: starrocks_catalog

Pekerjaan ingesti data dapat secara otomatis menggunakan kembali opsi katalog StarRocks berikut:

• jdbc-url

• http-url

• username

• password

• table.num-buckets

Untuk mengganti nilai-nilai ini, Anda dapat secara eksplisit mengatur opsi YAML yang sesuai, yang akan memiliki prioritas lebih tinggi.

Pemetaan tipe

Perubahan skema

Sebagai sink ingesti data, StarRocks mendukung event perubahan skema berikut:

• CREATE TABLE EVENT

  Catatan

  Jika tabel StarRocks hilir sudah ada, konektor tidak mencoba membuatnya lagi. Pastikan skema tabel hilir kompatibel dengan skema hulu.

• ADD COLUMN EVENT

  Catatan

  StarRocks mengharuskan kolom kunci primer muncul di awal tabel. Kolom baru apa pun harus ditambahkan setelahnya.

• ALTER COLUMN TYPE EVENT

  Catatan

  Untuk jalur perubahan skema yang didukung, lihat dokumentasi resmi StarRocks.

• DROP COLUMN EVENT

• TRUNCATE TABLE EVENT

• DROP TABLE EVENT

Sinkronisasi satu tabel

Sinkronkan satu tabel MySQL ke StarRocks. Jika database dan tabel tujuan tidak ada, konektor secara otomatis membuat tabel kunci primer.

pipeline:
          name: MySQL to StarRocks Pipeline
      source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
        # (Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental tanpa me-restart pekerjaan.
        scan.binlog.newly-added-table.enabled: true
        # (Opsional) Sinkronkan komentar tabel dan kolom ke tujuan.
        include-comments.enabled: true
        # (Opsional) Hanya deserialize binlog untuk tabel yang ditangkap untuk meningkatkan kinerja baca.
        scan.only.deserialize.captured.tables.changelog.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
       
        # (Opsional) Untuk pekerjaan dengan data sedikit, kurangi interval flush untuk mencegah penundaan persistensi lama. Default: 300000, atau 5 menit.
        sink.buffer-flush.interval-ms: 5000
        # (Opsional) Jika set karakter hulu adalah utf8mb4, atur parameter ini ke 4 untuk mencegah pemotongan teks. Default: 3.
        unicode-char.max-bytes: 4
        # (Opsional) Jumlah bucket untuk tabel yang dibuat otomatis. Parameter ini wajib untuk versi StarRocks sebelum 2.5.7. Versi yang lebih baru dapat menyimpulkan nilainya secara otomatis.
        table.create.num-buckets: 8
        # (Opsional) Jumlah replika untuk tabel yang dibuat otomatis. Konfigurasikan nilai ini berdasarkan kluster Anda.
        table.create.properties.replication_num: 3
        # (Opsional) Untuk StarRocks 3.2 dan yang lebih baru, aktifkan opsi ini untuk mempercepat perubahan skema.
        table.create.properties.fast_schema_evolution: true
        # Catatan: Jika Anda menggunakan transform untuk mengubah kunci primer, Anda juga harus mengatur sink.ignore.update-before: false.
        # Jika tidak, baris yang terkait dengan kunci primer lama tetap ada di tujuan.
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Sinkronisasi seluruh database

Sinkronkan semua tabel dalam database MySQL ke StarRocks secara bersamaan. Konektor secara otomatis membuat database tujuan dan tabel kunci primer, sehingga Anda tidak perlu membuat setiap tabel terlebih dahulu.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # Gunakan ekspresi reguler untuk mencocokkan semua tabel dalam database. Untuk mencocokkan beberapa database, pisahkan pola dengan koma.
        tables: test_db.\.*
        server-id: 5401-5499
        # (Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental tanpa me-restart pekerjaan.
        scan.binlog.newly-added-table.enabled: true
        # (Opsional) Sinkronkan komentar tabel dan kolom ke tujuan.
        include-comments.enabled: true
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Opsional) Untuk pekerjaan dengan data sedikit, kurangi interval flush untuk mencegah penundaan persistensi lama. Default: 300000, atau 5 menit.
        sink.buffer-flush.interval-ms: 5000
        # (Opsional) Jika set karakter hulu adalah utf8mb4, atur parameter ini ke 4 untuk mencegah pemotongan teks. Default: 3.
        unicode-char.max-bytes: 4
        # (Opsional) Jumlah bucket untuk tabel yang dibuat otomatis. Parameter ini wajib untuk versi StarRocks sebelum 2.5.7. Versi yang lebih baru dapat menyimpulkan nilainya secara otomatis.
        table.create.num-buckets: 8
        # (Opsional) Jumlah replika untuk tabel yang dibuat otomatis. Konfigurasikan nilai ini berdasarkan kluster Anda.
        table.create.properties.replication_num: 3
        # (Opsional) Untuk StarRocks 3.2 dan yang lebih baru, aktifkan opsi ini untuk mempercepat perubahan skema.
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Kecualikan tabel tertentu saat sinkronisasi database penuh

Saat menyinkronkan seluruh database, gunakan ekspresi reguler untuk melewati tabel yang tidak ingin Anda sinkronkan ke tujuan, seperti tabel sementara atau sensitif.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        # Tabel yang cocok dengan ekspresi reguler ini tidak disinkronkan.
        tables.exclude: test_db.tmp_.\*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Opsional) Versi antarmuka load. V2 memerlukan StarRocks 2.4 atau yang lebih baru. Untuk EMR Serverless StarRocks, gunakan V1 jika terjadi masalah streaming.
        sink.version: V2
        # (Opsional) Untuk pekerjaan dengan data sedikit, kurangi interval flush untuk mencegah penundaan persistensi lama. Default: 300000, atau 5 menit.
        sink.buffer-flush.interval-ms: 5000
        # (Opsional) Jumlah bucket untuk tabel yang dibuat otomatis. Parameter ini wajib untuk versi StarRocks sebelum 2.5.7.
        table.create.num-buckets: 8
        # (Opsional) Untuk StarRocks 3.2 dan yang lebih baru, aktifkan opsi ini untuk mempercepat perubahan skema.
        table.create.properties.fast_schema_evolution: true
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Sinkronisasi ke database dan tabel yang ditentukan

Jika nama database atau tabel StarRocks tujuan harus berbeda dari nama hulu, seperti saat menulis ke database lapisan ODS, gunakan route untuk mengganti namanya.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Opsional) Untuk pekerjaan dengan data sedikit, kurangi interval flush untuk mencegah penundaan persistensi lama. Default: 300000, atau 5 menit.
        sink.buffer-flush.interval-ms: 5000
        # (Opsional) Jumlah bucket untuk tabel yang dibuat otomatis. Parameter ini wajib untuk versi StarRocks sebelum 2.5.7.
        table.create.num-buckets: 8
      
      route:
        # Sinkronkan semua tabel dalam database test_db MySQL ke database test_db2 StarRocks tanpa mengubah nama tabel.
        # <> adalah placeholder yang diganti dengan nama tabel sumber yang cocok.
        - source-table: test_db.\.*
          sink-table: test_db2.<>
          replace-symbol: <>
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Gabungkan tabel terpartisi

Gabungkan beberapa tabel terpartisi dengan skema identik menjadi satu tabel StarRocks untuk kueri dan analitik terpadu.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        # Cocokkan semua tabel terpartisi, seperti user_0 dan user_1.
        tables: test_db.user\.*
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
        # (Opsional) Untuk pekerjaan dengan data sedikit, kurangi interval flush untuk mencegah penundaan persistensi lama. Default: 300000, atau 5 menit.
        sink.buffer-flush.interval-ms: 5000
        # (Opsional) Untuk tabel gabungan, tentukan secara eksplisit jumlah bucket berdasarkan volume data total.
        table.create.num-buckets: 8
      
      route:
        # Gabungkan semua tabel terpartisi ke tabel StarRocks test_db.user.
        - source-table: test_db.user\.*
          sink-table: test_db.user
      
      pipeline:
        name: MySQL to StarRocks Pipeline

Aktifkan mode EVOLVE

Secara default, mode LENIENT tidak menyinkronkan perubahan skema seperti menghapus kolom, menghapus tabel, atau memotong tabel ke tujuan. Jika Anda memerlukan sinkronisasi skema yang ketat, aktifkan mode EVOLVE. Mode ini memiliki batasan signifikan, jadi tinjau catatan berikut sebelum menggunakannya.

Batasan dan catatan penggunaan

• Penggantian nama kolom tidak didukung. Pekerjaan gagal jika terjadi event penggantian nama kolom hulu.

• Menghapus kolom atau tabel, atau memotong tabel, diterapkan ke tujuan. Operasi hulu yang tidak disengaja secara langsung memengaruhi tabel tujuan. Mode LENIENT default lebih aman karena tidak menyinkronkan event drop-table atau truncate-table.

• Jika Anda me-restart pekerjaan tanpa state dan tidak menghapus tabel sink, ketidakcocokan skema hulu dan sink dapat menyebabkan pekerjaan gagal. Anda harus menyesuaikan skema tabel hilir secara manual.

source:
        type: mysql
        name: MySQL Source
        hostname: <yourHostname>
        port: 3306
        username: <yourUsername>
        password: ${secret_values.mysql_password}
        tables: test_db.test_source_table
        server-id: 5401-5499
      
      sink:
        type: starrocks
        name: StarRocks Sink
        jdbc-url: jdbc:mysql://<yourFeHostname>:9030
        load-url: <yourFeHostname>:8030
        username: <yourUsername>
        password: ${secret_values.starrocks_password}
      
      pipeline:
        name: MySQL to StarRocks Pipeline
        # Aktifkan mode EVOLVE untuk menyinkronkan perubahan skema secara ketat. Pekerjaan gagal pada perubahan yang tidak didukung, seperti penggantian nama kolom.
        schema.change.behavior: evolve