E-MapReduce：Flink Connector

參數說明

資料類型映射

擷取Flink Connector JAR並上傳至Flink叢集

1. 您可以通過以下方式擷取Flink Connector JAR包。

   方式1：直接下載

   從Maven Central Repository擷取不同版本的Flink Connector JAR檔案。

   方式2：Maven依賴

   在Maven專案的pom.xml檔案中，根據以下格式將Flink Connector添加為依賴項。

   • 適用於Flink 1.15版本及以後的Flink Connector。

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

   • 適用於Flink 1.15版本之前的Flink Connector。

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

   方式3：手動編譯

   1. 下載Flink Connector代碼。

   2. 執行以下命令，將Flink Connector的原始碼編譯成一個JAR檔案。

      sh build.sh <flink_version>

      例如，如果您的環境中的Flink版本為1.17，您需要執行以下命令。

      sh build.sh 1.17

   3. 編譯完成後，在 target/ 目錄下找到產生的JAR檔案。

      例如，檔案名稱通常為flink-connector-starrocks-1.2.7_flink-1.17-SNAPSHOT.jar格式。

      說明

      非正式發布的Flink Connector版本會帶有SNAPSHOT尾碼。

   Flink Connector JAR檔案的命名格式如下：

   • 適用於Flink 1.15版本及以後的Flink Connector命名格式為flink-connector-starrocks-${connector_version}_flink-${flink_version}.jar。例如您安裝了Flink 1.17，並且想要使用1.2.8版本的Flink Connector，則您可以使用flink-connector-starrocks-1.2.8_flink-1.17.jar。

   • 適用於Flink 1.15版本之前的Flink Connector命名格式為flink-connector-starrocks-${connector_version}_flink-${flink_version}_${scala_version}.jar。例如您安裝了Flink 1.14和Scala 2.12，並且您想要使用1.2.7版本的Flink Connector，您可以使用flink-connector-starrocks-1.2.7_flink-1.14_2.12.jar。

     說明

     請根據實際情況替換以下資訊：

     • flink_version：Flink的版本號碼。

     • scala_version：Scala的版本號碼。

     • connector_version：Flink Connector的版本號碼。

2. 將擷取到的Flink Connector JAR檔案上傳至Flink叢集的flink-{flink_version}/lib目錄下。

   例如，如果您使用的是EMR叢集，且叢集版本為EMR-5.19.0，則JAR檔案應放置於/opt/apps/FLINK/flink-current/lib目錄下。

啟動Flink叢集

1. 登入Flink叢集的Master節點，詳情請參見登入叢集。

2. 執行以下命令，啟動Flink叢集。

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

使用Flink SQL寫入資料

1. 在StarRocks中建立一個名為test的資料庫，並在其中建立一張名為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)
   DISTRIBUTED BY HASH(id);

2. 登入Flink叢集的Master節點，詳情請參見登入叢集。

3. 執行以下命令，啟動Flink SQL。

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

4. 執行以下命令，建立一個名為score_board的表，並向其中插入資料。

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

   如果目標是將資料匯入StarRocks主鍵表，則必須在Flink表的DDL中明確指定主鍵。對於其他類型的StarRocks表（如Duplicate Key表），定義主鍵則為可選項。

使用Flink DataStream寫入資料

根據輸入記錄（input records）的不同類型，編寫對應Flink DataStream作業。

• 寫入CSV格式的字串資料

  如果輸入記錄為CSV格式的字串，對應的Flink DataStream作業的主要代碼如下所示，完整代碼請參見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);

• 寫入JSON格式的字串資料

  如果輸入記錄為JSON格式的字串，對應的Flink DataStream作業的主要代碼如下所示，完整代碼請參見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);

• 寫入自訂Java對象資料

  如果輸入記錄為自訂的Java對象，對應的Flink DataStream作業的主要代碼如下所示，完整代碼請參見LoadCustomJavaRecords。

  • 本樣本中，定義一個簡單的POJO類 RowData，用於表示每條記錄。

    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;
            }
        }
    

  • 主要代碼如下所示。

    // 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定義如下所示。

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

使用Flink CDC 3.0同步資料

Flink CDC 3.0架構能夠輕鬆構建從CDC資料來源（如MySQL、Kafka）到StarRocks的流式ELT管道。通過該管道，您可以實現以下功能：

• 自動建立資料庫和表

• 同步全量和增量資料

• 同步Schema Change

自StarRocks Flink Connector v1.2.9版本起，該連接器已整合至Flink CDC 3.0架構中，並被命名為StarRocks Pipeline Connector。該連接器具備上述所有功能，建議與StarRocks v3.2.1及以上版本配合使用，以充分利用fast_schema_evolution特性，進一步提升列的增減速度並降低資源消耗。

匯入至主鍵表

1. 在StarRocks中建立一個名為test的資料庫，並在其中建立一個主鍵表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. 向StarRocks表中插入資料。

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

3. 執行以下命令，啟動Flink SQL用戶端。

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

4. 更新資料。

   部分更新

   部分更新允許您僅更新指定列（如name），而不影響其他列（如score）。

   1. 在Flink SQL用戶端建立表score_board，並啟用部分更新功能。

      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：啟用部分更新。

      • sink.properties.columns：指定需要更新的列。如果 Flink Connector版本小於等於1.2.7，則還需要將選項sink.properties.columns設定為id,name,__op，以告訴 Flink connector 需要更新的列。請注意，您需要在末尾附加欄位__op。欄位__op表示匯入是 UPSERT 還是 DELETE 操作，其值由 Flink connector 自動化佈建。

   2. 插入更新資料。

      插入兩行資料，主鍵與現有資料相同，但name列的值被修改。

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

   3. 在SQL Editor中查詢StarRocks表。

      SELECT * FROM `test`.`score_board`;

      您會看到只有name列的值發生了變化，而score列保持不變。

      

   條件更新

   本樣本展示如何根據score列的值進行條件更新。只有匯入的資料行中score列值大於等於StarRocks表當前值時，該資料行才會更新。

   1. 在Flink SQL用戶端按照以下方式建立表score_board。

      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：設定為score，表示在資料寫入時，Flink Connector會以 score 列作為更新條件。

      • sink.version：設定為V1，表示Flink Connector使用Stream Load介面匯入資料。

   2. 在Flink SQL用戶端插入兩行資料到表中。

      資料行的主鍵與StarRocks表中的行相同。第一行資料score列中具有較小的值，而第二行資料score列中具有較大的值。

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

   3. 在SQL Editor中查詢StarRocks表。

      SELECT * FROM `test`.`score_board`;

      您會看到只有第二行資料發生了變化，而第一行資料未發生變化。

      

匯入至Bitmap列

Bitmap類型常用於加速精確去重計數情境，例如計算獨立訪客數（UV）。以下是一個完整的樣本，展示如何通過Flink SQL將資料匯入到StarRocks表的Bitmap列中，並在StarRocks中查詢UV數。

1. 在SQL Editor中建立StarRocks彙總表。

   在資料庫test中建立一個彙總表page_uv，其中：

   • visit_users列被定義為 BITMAP 類型，並配置彙總函式BITMAP_UNION。

   • page_id和visit_date作為彙總鍵（AGGREGATE KEY），用於分組和去重。

   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. 在Flink SQL用戶端中建立表。

   由於Flink不支援Bitmap類型，需要通過以下方式實現列映射和類型轉換：

   • 在Flink表中，將visit_user_id列定義為BIGINT類型，以代表StarRocks表中的visit_users列。

   • 使用sink.properties.columns配置，將visit_user_id列的資料通過to_bitmap函數轉換為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. 在Flink SQL用戶端中插入資料。

   向表page_uv插入多行資料，類比不同使用者在不同時間訪問頁面的行為。

   visit_user_id是BIGINT類型，Flink會將其自動轉換為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. 在SQL Editor中查詢UV數。

   通過StarRocks的彙總能力，使用COUNT(DISTINCT visit_users)計算每個頁面的獨立訪客數（UV）。

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

   返回結果如下所示。

   

匯入至HLL列

HLL（HyperLogLog）是一種用於近似去重計數的資料類型，適合處理大規模資料的獨立訪客數（UV）計算。以下是一個完整的樣本，展示如何通過Flink SQL將資料匯入到StarRocks表的HL 列中，並在StarRocks中查詢UV數。

1. 在SQL Editor中建立StarRocks彙總表。

   在資料庫test中建立一個彙總表hll_uv，其中：

   • visit_users列被定義為HLL類型，並配置彙總函式HLL_UNION。

   • page_id和visit_date作為彙總鍵（AGGREGATE KEY），用於分組和去重。

   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. 在Flink SQL用戶端中建立表。

   由於Flink不支援HLL類型，需要通過以下方式實現列映射和類型轉換：

   • 在Flink表中，將visit_user_id列定義為BIGINT類型，以代表StarRocks表中的visit_users列。

   • 使用sink.properties.columns配置列映射，並通過hll_hash函數將BIGINT類型的visit_user_id資料轉換為HLL類型。

   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. 在Flink SQL用戶端中插入資料。

   向表hll_uv插入多行資料，類比不同使用者在不同時間訪問頁面的行為。

   visit_user_id是BIGINT類型，Flink會將其自動轉換為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. 在SQL Editor中查詢UV數。

   通過StarRocks的彙總能力，使用COUNT(DISTINCT visit_users)計算每個頁面的獨立訪客數（UV）。

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

   返回結果如下所示。