E-MapReduce：Flink Connector

Parameter

Pemetaan tipe data

Mendapatkan JAR Flink Connector dan mengunggahnya ke kluster Flink

1. Anda dapat memperoleh paket JAR Flink Connector melalui metode berikut.

   Metode 1: Unduh langsung

   Ambil file JAR Flink Connector versi berbeda dari Repositori Maven Central.

   Metode 2: Dependensi Maven

   Dalam file pom.xml proyek Maven Anda, tambahkan Flink Connector sebagai dependensi menggunakan format berikut.

   • Untuk Flink Connector yang berlaku untuk Flink versi 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 berlaku untuk 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 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 di lingkungan Anda adalah 1.17, Anda perlu menjalankan perintah berikut.

      sh build.sh 1.17

   3. Setelah kompilasi selesai, temukan 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

      Rilis tidak resmi versi Flink Connector akan memiliki akhiran SNAPSHOT.

   Format penamaan file JAR Flink Connector adalah sebagai berikut:

   • Untuk Flink Connector yang berlaku untuk Flink versi 1.15 dan seterusnya, format penamaannya adalah flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar. Sebagai contoh, jika Anda telah menginstal Flink 1.17 dan ingin menggunakan Flink Connector versi 1.2.8, Anda dapat menggunakan flink-connector-starrocks-1.2.8_flink-1.17.jar.

   • Untuk Flink Connector yang berlaku untuk versi Flink sebelum 1.15, format penamaannya adalah flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar. Sebagai contoh, jika Anda telah menginstal Flink 1.14 dan Scala 2.12, serta ingin menggunakan Flink Connector versi 1.2.7, Anda dapat menggunakan flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar.

     Catatan

     Silakan ganti informasi berikut sesuai dengan situasi aktual Anda:

     • flink_version: Nomor versi Flink.

     • scala_version: Nomor versi Scala.

     • connector_version: Nomor versi Flink Connector.

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

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

Memulai kluster Flink

1. Masuk ke node master kluster Flink. Untuk informasi selengkapnya, lihat Masuk ke kluster.

2. Jalankan perintah berikut untuk memulai kluster Flink.

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

Menulis data menggunakan Flink SQL

1. Buat database bernama test di StarRocks, dan buat tabel kunci primer bernama score_board di dalamnya.

   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. Untuk informasi selengkapnya, lihat Masuk ke kluster.

3. Jalankan perintah berikut untuk memulai 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 tujuannya adalah mengimpor data ke tabel kunci primer StarRocks, Anda harus secara eksplisit menentukan kunci primer dalam DDL tabel Flink. Untuk jenis tabel StarRocks lainnya (seperti tabel Duplicate Key), mendefinisikan kunci primer bersifat opsional.

Menulis data menggunakan Flink DataStream

Tulis pekerjaan Flink DataStream yang sesuai berdasarkan jenis catatan input yang berbeda.

• Menulis data string dalam format CSV

  Jika catatan input berupa string dalam format CSV, kode utama pekerjaan Flink DataStream yang sesuai ditunjukkan di bawah ini. Untuk kode lengkapnya, lihat 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);

• Menulis data string dalam format JSON

  Jika catatan input berupa string dalam format JSON, kode utama pekerjaan Flink DataStream yang sesuai ditunjukkan di bawah ini. Untuk kode lengkapnya, lihat 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);

• Menulis data objek Java kustom

  Jika catatan input berupa objek Java kustom, kode utama pekerjaan Flink DataStream yang sesuai ditunjukkan di bawah ini. Untuk kode lengkapnya, lihat 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 ditunjukkan di bawah ini.

    // 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();
        }
    }  

Menyinkronkan data menggunakan Flink CDC 3.0

Framework Flink CDC 3.0 memungkinkan Anda dengan mudah membangun pipeline ELT streaming dari sumber data CDC (seperti MySQL, Kafka) ke StarRocks. Melalui pipeline ini, Anda dapat mencapai fungsi-fungsi berikut:

• Pembuatan database dan tabel secara otomatis

• Sinkronisasi data penuh dan inkremental

• Sinkronisasi Perubahan Skema

Mulai dari StarRocks Flink Connector v1.2.9, konektor ini telah diintegrasikan ke dalam framework Flink CDC 3.0 dan diberi nama StarRocks Pipeline Connector. Konektor ini memiliki semua fitur di atas dan direkomendasikan untuk digunakan dengan StarRocks v3.2.1 dan di atasnya agar dapat memanfaatkan sepenuhnya fitur fast_schema_evolution, yang lebih lanjut meningkatkan kecepatan penambahan dan penghapusan kolom serta mengurangi konsumsi sumber daya.

Impor ke tabel kunci primer

1. Buat database bernama test di StarRocks, dan buat tabel kunci primer score_board di dalamnya.

   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 memulai 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 lainnya (seperti score).

   1. Buat tabel score_board di klien Flink SQL dan aktifkan fitur 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 kurang dari atau sama dengan 1.2.7, Anda juga perlu mengatur opsi sink.properties.columns menjadi id,name,__op untuk memberi tahu konektor Flink kolom mana yang perlu diperbarui. Perhatikan bahwa Anda perlu menambahkan bidang __op di akhir. Bidang __op menunjukkan apakah impor tersebut merupakan operasi UPSERT atau DELETE, dan nilainya diatur secara otomatis oleh konektor Flink.

   2. Masukkan data pembaruan.

      Masukkan dua baris data dengan kunci primer yang sama seperti data yang sudah ada, tetapi dengan nilai yang dimodifikasi pada 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 pada kolom name yang berubah, sedangkan kolom score tetap tidak berubah.

      

   Pembaruan bersyarat

   Contoh ini menunjukkan cara melakukan pembaruan bersyarat berdasarkan nilai kolom score. Baris data diperbarui hanya ketika nilai kolom score dalam data yang diimpor 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, menunjukkan bahwa saat data ditulis, Flink Connector akan menggunakan kolom score sebagai kondisi pembaruan.

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

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

      Baris data memiliki kunci primer yang sama dengan baris di tabel StarRocks. Baris data pertama memiliki nilai yang lebih kecil pada kolom score, sedangkan baris data kedua memiliki nilai yang lebih besar pada kolom score.

      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 data kedua yang berubah, sedangkan baris data pertama tetap tidak berubah.

      

Impor ke kolom Bitmap

Tipe Bitmap umumnya digunakan untuk mempercepat skenario penghitungan deduplikasi eksak, seperti menghitung pengunjung unik (UV). Berikut adalah contoh lengkap yang menunjukkan cara mengimpor data ke kolom Bitmap dalam tabel StarRocks melalui 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, di mana:

   • 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, pemetaan kolom dan konversi tipe perlu diimplementasikan sebagai berikut:

   • Di tabel Flink, definisikan kolom visit_user_id sebagai tipe 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 melalui 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 data ke tabel page_uv, mensimulasikan pengguna berbeda yang mengakses halaman pada waktu berbeda.

   visit_user_id bertipe BIGINT, dan Flink akan 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) untuk setiap halaman menggunakan COUNT(DISTINCT visit_users).

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

   Hasil yang dikembalikan ditunjukkan di bawah ini.

   

Impor ke kolom HLL

HLL (HyperLogLog) adalah tipe data yang digunakan untuk penghitungan deduplikasi perkiraan, cocok untuk menghitung pengunjung unik (UV) dalam skenario data berskala besar. Berikut adalah contoh lengkap yang menunjukkan cara mengimpor data ke kolom HLL dalam tabel StarRocks melalui 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, di mana:

   • 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, pemetaan kolom dan konversi tipe perlu diimplementasikan sebagai berikut:

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

   • Gunakan konfigurasi sink.properties.columns untuk pemetaan kolom, dan konversikan data tipe BIGINT visit_user_id ke tipe HLL melalui 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 data ke tabel hll_uv, mensimulasikan pengguna berbeda yang mengakses halaman pada waktu berbeda.

   visit_user_id bertipe BIGINT, dan Flink akan 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) untuk setiap halaman menggunakan COUNT(DISTINCT visit_users).

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

   Hasil yang dikembalikan ditunjukkan di bawah ini.