E-MapReduce:Flink Connector

Deskripsi parameter

Pemetaan tipe data

Peroleh file JAR Flink Connector dan unggah ke kluster Flink Anda

1. Anda dapat memperoleh file JAR Flink Connector dengan salah satu metode berikut.

   Metode 1: Unduh langsung

   Anda dapat mengunduh berbagai versi file JAR Flink Connector dari Maven Central Repository.

   Metode 2: Dependensi Maven

   Tambahkan Flink Connector sebagai dependensi ke file pom.xml proyek Maven Anda dalam format berikut.

   • Untuk Flink Connector yang kompatibel dengan Flink 1.15 dan seterusnya:

     <dependency>
         <groupId>com.starrocks</groupId>
         <artifactId>flink-connector-starrocks</artifactId>
         <version>${connector_version}_flink-${flink_version}</version>
     </dependency>

   • Untuk Flink Connector yang kompatibel dengan versi Flink sebelum 1.15:

     <dependency>
         <groupId>com.starrocks</groupId>
         <artifactId>flink-connector-starrocks</artifactId>
         <version>${connector_version}_flink-${flink_version}_${scala_version}</version>
     </dependency>

   Metode 3: Kompilasi manual

   1. Unduh kode sumber Flink Connector.

   2. Jalankan perintah berikut untuk mengompilasi kode sumber Flink Connector menjadi file JAR.

      sh build.sh <flink_version>

      Sebagai contoh, jika versi Flink Anda adalah 1.17, jalankan perintah berikut.

      sh build.sh 1.17

   3. Setelah kompilasi, Anda dapat menemukan file JAR yang dihasilkan di direktori target/.

      Sebagai contoh, nama file biasanya dalam format flink-connector-starrocks-1.2.7_flink-1.17-SNAPSHOT.jar.

      Catatan

      Versi Flink Connector yang belum dirilis mencakup akhiran SNAPSHOT.

   Nama file JAR Flink Connector mengikuti format berikut:

   • Untuk Flink 1.15 dan seterusnya: flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar. Sebagai contoh, jika Anda menggunakan Flink 1.17 dan Flink Connector 1.2.8, nama file-nya adalah flink-connector-starrocks-1.2.8_flink-1.17.jar.

   • Untuk versi Flink sebelum 1.15: flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar. Sebagai contoh, jika Anda menggunakan Flink 1.14, Scala 2.12, dan Flink Connector 1.2.7, nama file-nya adalah flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar.

     Catatan

     Ganti placeholder berikut sesuai kebutuhan:

     • flink_version: Nomor versi Flink Anda.

     • scala_version: Nomor versi Scala Anda.

     • connector_version: Nomor versi Flink Connector Anda.

2. Unggah file JAR Flink Connector yang telah Anda peroleh ke direktori flink-{flink_version}/lib kluster Flink Anda.

   Sebagai contoh, jika Anda menggunakan kluster EMR versi EMR-5.19.0, letakkan file JAR di direktori /opt/apps/FLINK/flink-current/lib.

Jalankan kluster Flink

1. Masuk ke node master kluster Flink Anda. Untuk informasi lebih lanjut, lihat Masuk ke kluster.

2. Jalankan perintah berikut untuk menjalankan kluster Flink.

   /opt/apps/FLINK/flink-current/bin/start-cluster.sh

Tulis data menggunakan Flink SQL

1. Buat database bernama test di StarRocks, lalu buat tabel primary key bernama score_board di database tersebut.

   CREATE DATABASE test;
   CREATE TABLE test.score_board(
       id int(11) NOT NULL COMMENT "",
       name varchar(65533) NULL DEFAULT "" COMMENT "",
       score int(11) NOT NULL DEFAULT "0" COMMENT ""
   )
   ENGINE=OLAP
   PRIMARY KEY(id)
   DISTRIBUTED BY HASH(id);

2. Masuk ke node master kluster Flink Anda. Untuk informasi lebih lanjut, lihat Masuk ke kluster.

3. Jalankan perintah berikut untuk menjalankan Flink SQL.

   /opt/apps/FLINK/flink-current/bin/sql-client.sh

4. Jalankan perintah berikut untuk membuat tabel bernama score_board dan memasukkan data ke dalamnya.

   CREATE TABLE `score_board` (
       `id` INT,
       `name` STRING,
       `score` INT,
       PRIMARY KEY (id) NOT ENFORCED
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'score_board',
       'username' = 'admin',
       'password' = '<password>',
   );
   INSERT INTO `score_board` VALUES (1, 'starrocks', 100), (2, 'flink', 100);

   Jika Anda mengimpor data ke tabel primary key StarRocks, Anda harus secara eksplisit menentukan primary key dalam DDL tabel Flink. Untuk jenis tabel StarRocks lainnya, seperti tabel Duplicate Key, menentukan primary key bersifat opsional.

Tulis data menggunakan Flink DataStream

Tulis pekerjaan Flink DataStream berdasarkan jenis catatan input.

• Tulis data string berformat CSV

  Jika catatan input Anda berupa string berformat CSV, lihat kode lengkap pekerjaan Flink DataStream di LoadCsvRecords.

  /**
   * Generate CSV-format records. Each record has three values separated by "\t". 
   * These values will be loaded to the columns `id`, `name`, and `score` in the StarRocks table.
   */
  String[] records = new String[]{
          "1\tstarrocks-csv\t100",
          "2\tflink-csv\t100"
  };
  DataStream<String> source = env.fromElements(records);
  /**
   * Configure the Flink connector with the required properties.
   * You also need to add properties "sink.properties.format" and "sink.properties.column_separator"
   * to tell the Flink connector the input records are CSV-format, and the column separator is "\t".
   * You can also use other column separators in the CSV-format records,
   * but remember to modify the "sink.properties.column_separator" correspondingly.
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "root")
          .withProperty("password", "")
          .withProperty("sink.properties.format", "csv")
          .withProperty("sink.properties.column_separator", "\t")
          .build();
  // Create the sink with the options.
  SinkFunction<String> starRockSink = StarRocksSink.sink(options);
  source.addSink(starRockSink);

• Tulis data string berformat JSON

  Jika catatan input Anda berupa string berformat JSON, lihat kode lengkap pekerjaan Flink DataStream di LoadJsonRecords.

  /**
   * Generate JSON-format records. 
   * Each record has three key-value pairs corresponding to the columns id, name, and score in the StarRocks table.
   */
  String[] records = new String[]{
          "{\"id\":1, \"name\":\"starrocks-json\", \"score\":100}",
          "{\"id\":2, \"name\":\"flink-json\", \"score\":100}",
  };
  DataStream<String> source = env.fromElements(records);
  /** 
   * Configure the Flink connector with the required properties.
   * You also need to add properties "sink.properties.format" and "sink.properties.strip_outer_array"
   * to tell the Flink connector the input records are JSON-format and to strip the outermost array structure. 
   */
  StarRocksSinkOptions options = StarRocksSinkOptions.builder()
          .withProperty("jdbc-url", jdbcUrl)
          .withProperty("load-url", loadUrl)
          .withProperty("database-name", "test")
          .withProperty("table-name", "score_board")
          .withProperty("username", "root")
          .withProperty("password", "")
          .withProperty("sink.properties.format", "json")
          .withProperty("sink.properties.strip_outer_array", "true")
          .build();
  // Create the sink with the options.
  SinkFunction<String> starRockSink = StarRocksSink.sink(options);
  source.addSink(starRockSink);

• Tulis data objek Java kustom

  Jika catatan input Anda berupa objek Java kustom, lihat kode lengkap pekerjaan Flink DataStream di LoadCustomJavaRecords.

  • Dalam contoh ini, kelas POJO sederhana RowData didefinisikan untuk merepresentasikan setiap catatan.

    public static class RowData {
            public int id;
            public String name;
            public int score;
            public RowData() {}
            public RowData(int id, String name, int score) {
                this.id = id;
                this.name = name;
                this.score = score;
            }
        }
    

  • Kode utamanya adalah sebagai berikut.

    // Generate records which use RowData as the container.
    RowData[] records = new RowData[]{
            new RowData(1, "starrocks-rowdata", 100),
            new RowData(2, "flink-rowdata", 100),
        };
    DataStream<RowData> source = env.fromElements(records);
    // Configure the Flink connector with the required properties.
    StarRocksSinkOptions options = StarRocksSinkOptions.builder()
            .withProperty("jdbc-url", jdbcUrl)
            .withProperty("load-url", loadUrl)
            .withProperty("database-name", "test")
            .withProperty("table-name", "score_board")
            .withProperty("username", "root")
            .withProperty("password", "")
            .build();
    /**
     * The Flink connector will use a Java object array (Object[]) to represent a row to be loaded into the StarRocks table,
     * and each element is the value for a column.
     * You need to define the schema of the Object[] which matches that of the StarRocks table.
     */
    TableSchema schema = TableSchema.builder()
            .field("id", DataTypes.INT().notNull())
            .field("name", DataTypes.STRING())
            .field("score", DataTypes.INT())
            // When the StarRocks table is a Primary Key table, you must specify notNull(), for example, DataTypes.INT().notNull(), for the primary key `id`.
            .primaryKey("id")
            .build();
    // Transform the RowData to the Object[] according to the schema.
    RowDataTransformer transformer = new RowDataTransformer();
    // Create the sink with the schema, options, and transformer.
    SinkFunction<RowData> starRockSink = StarRocksSink.sink(schema, options, transformer);
    source.addSink(starRockSink);
    

    RowDataTransformer didefinisikan sebagai berikut.

    private static class RowDataTransformer implements StarRocksSinkRowBuilder<RowData> {
        /**
         * Set each element of the object array according to the input RowData.
         * The schema of the array matches that of the StarRocks table.
         */
        @Override
        public void accept(Object[] internalRow, RowData rowData) {
            internalRow[0] = rowData.id;
            internalRow[1] = rowData.name;
            internalRow[2] = rowData.score;
            // When the StarRocks table is a Primary Key table, you need to set the last element to indicate whether the data loading is an UPSERT or DELETE operation.
            internalRow[internalRow.length - 1] = StarRocksSinkOP.UPSERT.ordinal();
        }
    }  

Sinkronkan data menggunakan Flink CDC 3.0

Framework Flink CDC 3.0 menyederhanakan pembuatan pipeline ELT streaming dari sumber data CDC, seperti MySQL atau Kafka, ke StarRocks. Dengan pipeline ini, Anda dapat:

• Membuat database dan tabel secara otomatis

• Menyinkronkan data penuh dan inkremental

• Menyinkronkan perubahan skema

Mulai dari StarRocks Flink Connector v1.2.9, connector ini diintegrasikan ke dalam framework Flink CDC 3.0 dan diganti namanya menjadi StarRocks Pipeline Connector. Connector ini mendukung semua fitur di atas. Kami merekomendasikan Anda menggunakannya dengan StarRocks v3.2.1 atau lebih baru untuk memanfaatkan fitur `fast_schema_evolution`, yang lebih mempercepat penambahan dan penghapusan kolom serta mengurangi konsumsi sumber daya.

Impor ke tabel primary key

1. Buat database bernama test di StarRocks, lalu buat tabel primary key bernama score_board.

   CREATE DATABASE `test`;
   CREATE TABLE `test`.`score_board`
   (
       `id` int(11) NOT NULL COMMENT "",
       `name` varchar(65533) NULL DEFAULT "" COMMENT "",
       `score` int(11) NOT NULL DEFAULT "0" COMMENT ""
   )
   ENGINE=OLAP
   PRIMARY KEY(`id`)
   COMMENT "OLAP"
   DISTRIBUTED BY HASH(`id`);

2. Masukkan data ke tabel StarRocks.

   INSERT INTO `test`.`score_board` VALUES (1, 'starrocks', 100), (2, 'flink', 100);

3. Jalankan perintah berikut untuk menjalankan klien Flink SQL.

   /opt/apps/FLINK/flink-current/bin/sql-client.sh

4. Perbarui data.

   Pembaruan parsial

   Pembaruan parsial memungkinkan Anda memperbarui hanya kolom tertentu, seperti name, tanpa memengaruhi kolom lain, seperti score.

   1. Buat tabel score_board di klien Flink SQL dan aktifkan pembaruan parsial.

      CREATE TABLE `score_board` (
          `id` INT,
          `name` STRING,
          PRIMARY KEY (id) NOT ENFORCED
      ) WITH (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
          'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
          'database-name' = 'test',
          'table-name' = 'score_board',
          'username' = 'admin',
          'password' = '<password>',
          'sink.properties.partial_update' = 'true',
          -- only for Flink connector version <= 1.2.7
          'sink.properties.columns' = 'id,name,__op'
      ); 

      • sink.properties.partial_update: Mengaktifkan pembaruan parsial.

      • sink.properties.columns: Menentukan kolom yang akan diperbarui. Jika versi Flink Connector adalah 1.2.7 atau lebih lama, Anda juga harus mengatur opsi sink.properties.columns ke id,name,__op untuk memberi tahu Flink connector kolom mana yang akan diperbarui. Perhatikan bahwa Anda harus menambahkan bidang __op di akhir. Bidang __op menunjukkan apakah impor merupakan operasi UPSERT atau DELETE, dan nilainya diatur secara otomatis oleh Flink connector.

   2. Masukkan data yang diperbarui.

      Masukkan dua baris yang memiliki primary key sama dengan data yang ada tetapi dengan nilai yang dimodifikasi di kolom name.

      INSERT INTO score_board VALUES (1, 'starrocks-update'), (2, 'flink-update');

   3. Kueri tabel StarRocks di SQL Editor.

      SELECT * FROM `test`.`score_board`;

      Anda akan melihat bahwa hanya nilai di kolom name yang berubah, sedangkan kolom score tetap tidak berubah.

      Kueri mengembalikan dua catatan: id=1, name=starrocks-update, score=100 dan id=2, name=flink-update, score=100.

   Pembaruan bersyarat

   Contoh ini menunjukkan cara melakukan pembaruan bersyarat berdasarkan nilai kolom score. Baris diperbarui hanya jika nilai score pada data masuk lebih besar dari atau sama dengan nilai saat ini di tabel StarRocks.

   1. Buat tabel score_board di klien Flink SQL sebagai berikut.

      CREATE TABLE `score_board` (
          `id` INT,
          `name` STRING,
          `score` INT,
          PRIMARY KEY (id) NOT ENFORCED
      ) WITH (
          'connector' = 'starrocks',
          'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
          'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
          'database-name' = 'test',
          'table-name' = 'score_board',
          'username' = 'admin',
          'password' = '<password>',
          'sink.properties.merge_condition' = 'score',
          'sink.version' = 'V1'
      );

      • sink.properties.merge_condition: Diatur ke score. Ini menentukan bahwa Flink Connector menggunakan kolom score sebagai kondisi pembaruan selama penulisan data.

      • sink.version: Diatur ke V1. Ini menentukan bahwa Flink Connector menggunakan antarmuka Stream Load untuk mengimpor data.

   2. Masukkan dua baris data ke tabel di klien Flink SQL.

      Baris-baris tersebut memiliki primary key yang sama dengan baris yang ada di tabel StarRocks. Baris pertama memiliki nilai score lebih kecil, dan baris kedua memiliki nilai score lebih besar.

      INSERT INTO `score_board` VALUES (1, 'starrocks-update', 99), (2, 'flink-update', 101);

   3. Kueri tabel StarRocks di SQL Editor.

      SELECT * FROM `test`.`score_board`;

      Anda akan melihat bahwa hanya baris kedua yang berubah, sedangkan baris pertama tetap tidak berubah.

      Kueri mengembalikan dua catatan: baris pertama id=1, name=starrocks, score=100 (tidak berubah) dan baris kedua id=2, name=flink-update, score=101 (diperbarui).

Impor ke kolom Bitmap

Tipe data Bitmap umumnya digunakan untuk mempercepat skenario penghitungan jumlah unik yang tepat, seperti menghitung pengunjung unik (UV). Contoh berikut menunjukkan cara mengimpor data ke kolom Bitmap tabel StarRocks menggunakan Flink SQL dan mengkueri jumlah UV di StarRocks.

1. Buat tabel agregat StarRocks di SQL Editor.

   Buat tabel agregat bernama page_uv di database test dengan spesifikasi berikut:

   • Kolom visit_users didefinisikan sebagai tipe BITMAP dan dikonfigurasi dengan fungsi agregat `BITMAP_UNION`.

   • page_id dan visit_date berfungsi sebagai kunci agregat (AGGREGATE KEY) untuk pengelompokan dan deduplikasi.

   CREATE TABLE `test`.`page_uv` (
     `page_id` INT NOT NULL COMMENT 'page ID',
     `visit_date` datetime NOT NULL COMMENT 'access time',
     `visit_users` BITMAP BITMAP_UNION NOT NULL COMMENT 'user ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. Buat tabel di klien Flink SQL.

   Karena Flink tidak mendukung tipe Bitmap, Anda harus menerapkan pemetaan kolom dan konversi tipe sebagai berikut:

   • Di tabel Flink, definisikan kolom visit_user_id sebagai BIGINT untuk merepresentasikan kolom visit_users di tabel StarRocks.

   • Gunakan konfigurasi sink.properties.columns untuk mengonversi data di kolom visit_user_id ke tipe Bitmap menggunakan fungsi `to_bitmap`.

   CREATE TABLE `page_uv` (
       `page_id` INT,
       `visit_date` TIMESTAMP,
       `visit_user_id` BIGINT
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'page_uv',
       'username' = 'admin',
       'password' = '<password>',
       'sink.properties.columns' = 'page_id,visit_date,visit_user_id,visit_users=to_bitmap(visit_user_id)'
   );

3. Masukkan data di klien Flink SQL.

   Masukkan beberapa baris ke tabel page_uv untuk mensimulasikan pengguna berbeda mengakses halaman pada waktu berbeda.

   visit_user_id bertipe BIGINT. Flink secara otomatis mengonversinya ke tipe Bitmap.

   INSERT INTO `page_uv` VALUES
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 13),
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 23),
      (1, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 33),
      (1, CAST('2020-06-23 02:30:30' AS TIMESTAMP), 13),
      (2, CAST('2020-06-23 01:30:30' AS TIMESTAMP), 23);

4. Kueri jumlah UV di SQL Editor.

   Gunakan kemampuan agregasi StarRocks untuk menghitung jumlah pengunjung unik (UV) per halaman dengan COUNT(DISTINCT visit_users).

   SELECT page_id, COUNT(DISTINCT visit_users) FROM page_uv GROUP BY page_id;

   Untuk page_id 2, count(DISTINCT visit_users) adalah 1. Untuk page_id 1, count(DISTINCT visit_users) adalah 3.

Impor ke kolom HLL

HyperLogLog (HLL) adalah tipe data yang digunakan untuk penghitungan jumlah unik perkiraan dan cocok untuk menghitung pengunjung unik (UV) dalam skenario data skala besar. Contoh berikut menunjukkan cara mengimpor data ke kolom HLL tabel StarRocks menggunakan Flink SQL dan mengkueri jumlah UV di StarRocks.

1. Buat tabel agregat StarRocks di SQL Editor.

   Buat tabel agregat bernama hll_uv di database test dengan spesifikasi berikut:

   • Kolom visit_users didefinisikan sebagai tipe HLL dan dikonfigurasi dengan fungsi agregat `HLL_UNION`.

   • page_id dan visit_date berfungsi sebagai kunci agregat (AGGREGATE KEY) untuk pengelompokan dan deduplikasi.

   CREATE TABLE `test`.`hll_uv` (
     `page_id` INT NOT NULL COMMENT 'page ID',
     `visit_date` DATETIME NOT NULL COMMENT 'access time',
     `visit_users` HLL HLL_UNION NOT NULL COMMENT 'user ID'
   ) ENGINE=OLAP
   AGGREGATE KEY(`page_id`, `visit_date`)
   DISTRIBUTED BY HASH(`page_id`);

2. Buat tabel di klien Flink SQL.

   Karena Flink tidak mendukung tipe HLL, Anda harus menerapkan pemetaan kolom dan konversi tipe sebagai berikut:

   • Di tabel Flink, definisikan kolom visit_user_id sebagai BIGINT untuk merepresentasikan kolom visit_users di tabel StarRocks.

   • Gunakan konfigurasi sink.properties.columns untuk memetakan kolom dan mengonversi data visit_user_id bertipe BIGINT ke tipe HLL menggunakan fungsi `hll_hash`.

   CREATE TABLE `hll_uv` (
       `page_id` INT,
       `visit_date` TIMESTAMP,
       `visit_user_id` BIGINT
   ) WITH (
       'connector' = 'starrocks',
       'jdbc-url' = 'jdbc:mysql://<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:9030',
       'load-url' = '<fe-{srClusterId}-internal.starrocks.aliyuncs.com>:8030',
       'database-name' = 'test',
       'table-name' = 'hll_uv',
       'username' = 'admin',
       'password' = '<password>',
       'sink.properties.columns' = 'page_id,visit_date,visit_user_id,visit_users=hll_hash(visit_user_id)'
   );

3. Masukkan data di klien Flink SQL.

   Masukkan beberapa baris ke tabel hll_uv untuk mensimulasikan pengguna berbeda mengakses halaman pada waktu berbeda.

   visit_user_id bertipe BIGINT. Flink secara otomatis mengonversinya ke tipe HLL.

   INSERT INTO `hll_uv` VALUES
      (3, CAST('2023-07-24 12:00:00' AS TIMESTAMP), 78),
      (4, CAST('2023-07-24 13:20:10' AS TIMESTAMP), 2),
      (3, CAST('2023-07-24 12:30:00' AS TIMESTAMP), 674);

4. Kueri jumlah UV di SQL Editor.

   Gunakan kemampuan agregasi StarRocks untuk menghitung jumlah pengunjung unik (UV) per halaman dengan COUNT(DISTINCT visit_users).

   SELECT `page_id`, COUNT(DISTINCT `visit_users`) FROM `hll_uv` GROUP BY `page_id`;

   Hasilnya adalah sebagai berikut.

     page_id  count(DISTINCT visit_users)
   0       4                            1
   1       3                            2