E-MapReduce:Flink Connector

Metode 1: DataStream API

// -------- sink dengan aliran string JSON mentah --------
fromElements(new String[]{
    "{\"score\": \"99\", \"name\": \"stephen\"}",
    "{\"score\": \"100\", \"name\": \"lebron\"}"
}).addSink(
    StarRocksSink.sink(
        // opsi sink
        StarRocksSinkOptions.builder()
            .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
            .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
            .withProperty("username", "xxx")
            .withProperty("password", "xxx")
            .withProperty("table-name", "xxx")
            .withProperty("database-name", "xxx")
            .withProperty("sink.properties.format", "json")
            .withProperty("sink.properties.strip_outer_array", "true")
            .build()
    )
);


// -------- sink dengan transformasi aliran --------
class RowData {
    public int score;
    public String name;
    public RowData(int score, String name) {
        ......
    }
}
fromElements(
    new RowData[]{
        new RowData(99, "stephen"),
        new RowData(100, "lebron")
    }
).addSink(
    StarRocksSink.sink(
        // struktur tabel
        TableSchema.builder()
            .field("score", DataTypes.INT())
            .field("name", DataTypes.VARCHAR(20))
            .build(),
        // opsi sink
        StarRocksSinkOptions.builder()
            .withProperty("jdbc-url", "jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx")
            .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
            .withProperty("username", "xxx")
            .withProperty("password", "xxx")
            .withProperty("table-name", "xxx")
            .withProperty("database-name", "xxx")
            .withProperty("sink.properties.column_separator", "\\x01")
            .withProperty("sink.properties.row_delimiter", "\\x02")
            .build(),
        // atur slot dengan streamRowData
        (slots, streamRowData) -> {
            slots[0] = streamRowData.score;
            slots[1] = streamRowData.name;
        }
    )
);

Metode 2: Pembuatan tabel SQL

Sebelum menjalankan metode ini, tambahkan com.starrocks.connector.flink.table.StarRocksDynamicTableSinkFactory ke file src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory.

// buat tabel dengan `struktur` dan `properti`
tEnv.executeSql(
    "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'," +
        "'sink.buffer-flush.max-rows' = '1000000'," +
        "'sink.buffer-flush.max-bytes' = '300000000'," +
        "'sink.buffer-flush.interval-ms' = '5000'," +
        "'sink.properties.column_separator' = '\\x01'," +
        "'sink.properties.row_delimiter' = '\\x02'," +
        "'sink.max-retries' = '3'" +
        "'sink.properties.*' = 'xxx'" + // properti stream load seperti `'sink.properties.columns' = 'k1, v1'`
    ")"
);

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

• Apache Flink 1.13 (disarankan) atau 1.11 / 1.12

• Akses ke database MySQL dan kluster StarRocks

Siapkan lingkungan

1. Unduh Apache Flink. Disarankan menggunakan Apache Flink 1.13. Versi paling awal yang didukung adalah 1.11.

2. Unduh connector Flink CDC untuk MySQL. Pilih paket yang sesuai dengan versi Flink Anda.

3. Unduh connector Flink StarRocks.

   Penting

   Connector Flink StarRocks berbeda untuk Apache Flink 1.13, 1.11, dan 1.12. Unduh versi yang tepat.

4. Salin kedua file flink-sql-connector-mysql-cdc-xxx.jar dan flink-connector-starrocks-xxx.jar ke direktori flink-xxx/lib/.

5. Unduh paket smt.tar.gz, ekstrak, lalu edit file konfigurasi dengan detail koneksi Anda. File konfigurasi menggunakan bagian-bagian berikut: Contoh konfigurasi:

   • db — informasi koneksi database MySQL

   • be_num — jumlah node dalam kluster StarRocks

   • [table-rule.N] — aturan pencocokan menggunakan ekspresi reguler untuk nama database dan nama tabel. Konfigurasikan satu bagian per set aturan.

   • flink.starrocks.* — konfigurasi kluster StarRocks

   [db]
   host = 192.168.**.**
   port = 3306
   user = root
   password =
   
   [other]
   # jumlah backend di StarRocks
   be_num = 3
   # `decimal_v3` didukung sejak StarRocks-1.18.1
   use_decimal_v3 = false
   # file untuk menyimpan DDL SQL hasil konversi
   output_dir = ./result
   
   [table-rule.1]
   # pola untuk mencocokkan database guna mengatur properti
   database = ^console_19321.*$
   # pola untuk mencocokkan tabel guna mengatur properti
   table = ^.*$
   
   ############################################
   ### konfigurasi sink flink
   ### JANGAN atur `connector`, `table-name`, `database-name`, karena di-generate otomatis
   ############################################
   flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
   flink.starrocks.load-url= 192.168.**.**:8030
   flink.starrocks.username=root
   flink.starrocks.password=
   flink.starrocks.sink.properties.column_separator=\x01
   flink.starrocks.sink.properties.row_delimiter=\x02
   flink.starrocks.sink.buffer-flush.interval-ms=15000

Buat tabel dan mulai sinkronisasi

1. Jalankan starrocks-migrate-tool untuk menghasilkan pernyataan CREATE TABLE. Output ditulis ke direktori /result. Jalankan ls result untuk memverifikasi file yang dihasilkan:

   flink-create.1.sql    smt.tar.gz              starrocks-create.all.sql
   flink-create.all.sql  starrocks-create.1.sql

2. Buat database dan tabel di StarRocks:

   Mysql -h <alamat IP frontend> -P 9030 -u root -p < starrocks-create.1.sql

3. Buat tabel Flink dan mulai sinkronisasi data berkelanjutan:

   Penting

   Jika Anda menggunakan Apache Flink versi sebelum 1.13, Anda tidak dapat menjalankan skrip SQL secara langsung. Eksekusi pernyataan SQL satu per satu dan aktifkan binary logging untuk database MySQL.

   bin/sql-client.sh -f flink-create.1.sql

4. Periksa status pekerjaan Flink yang sedang berjalan:

   bin/flink list -running

   Lihat detail dan status pekerjaan di UI web Flink atau di file log di bawah $FLINK_HOME/log.

Konfigurasi beberapa set aturan

Untuk menyinkronkan beberapa database atau tabel dengan properti berbeda, tambahkan satu bagian [table-rule.N] per set aturan. Konfigurasikan parameter flink.starrocks.sink secara independen untuk setiap set aturan, misalnya frekuensi impor yang berbeda.

Contoh konfigurasi dengan dua set aturan:

[table-rule.1]
# pola untuk mencocokkan database guna mengatur properti
database = ^console_19321.*$
# pola untuk mencocokkan tabel guna mengatur properti
table = ^.*$

############################################
### konfigurasi sink flink
### JANGAN atur `connector`, `table-name`, `database-name`, karena di-generate otomatis
############################################
flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
flink.starrocks.load-url= 192.168.**.**:8030
flink.starrocks.username=root
flink.starrocks.password=
flink.starrocks.sink.properties.column_separator=\x01
flink.starrocks.sink.properties.row_delimiter=\x02
flink.starrocks.sink.buffer-flush.interval-ms=15000

[table-rule.2]
# pola untuk mencocokkan database guna mengatur properti
database = ^database2.*$
# pola untuk mencocokkan tabel guna mengatur properti
table = ^.*$

############################################
### konfigurasi sink flink
### JANGAN atur `connector`, `table-name`, `database-name`, karena di-generate otomatis
############################################
flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
flink.starrocks.load-url= 192.168.**.**:8030
flink.starrocks.username=root
flink.starrocks.password=
# Jika Anda tidak dapat memilih pemisah CSV yang sesuai, alihkan ke format JSON.
# Perhatikan bahwa performa impor JSON mungkin menurun dibandingkan CSV.
flink.starrocks.sink.properties.strip_outer_array=true
flink.starrocks.sink.properties.format=json

Konsolidasi tabel yang di-shard

Setelah sharding horizontal, data dari tabel besar mungkin terbagi di beberapa tabel atau database. Konfigurasikan satu set aturan untuk menyinkronkan beberapa tabel sumber ke satu tabel StarRocks.

Sebagai contoh, jika edu_db_1 dan edu_db_2 masing-masing berisi course_1 dan course_2 dengan skema yang sama, aturan berikut mengonsolidasi keempat tabel tersebut ke satu tabel StarRocks:

[table-rule.3]
# pola untuk mencocokkan database guna mengatur properti
database = ^edu_db_[0-9]*$
# pola untuk mencocokkan tabel guna mengatur properti
table = ^course_[0-9]*$

############################################
### konfigurasi sink flink
### JANGAN atur `connector`, `table-name`, `database-name`, karena di-generate otomatis
############################################
flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
flink.starrocks.load-url= 192.168.**.**:8030
flink.starrocks.username=root
flink.starrocks.password=
flink.starrocks.sink.properties.column_separator=\x01
flink.starrocks.sink.properties.row_delimiter=\x02
flink.starrocks.sink.buffer-flush.interval-ms=5000

Setelah menjalankan starrocks-migrate-tool, hubungan sinkronisasi many-to-one dibuat secara otomatis. Tabel StarRocks yang dihasilkan diberi nama course__auto_shard secara default. Untuk mengganti namanya, edit file SQL yang dihasilkan sebelum menjalankannya.