E-MapReduce:Flink connector

Add the following code to the pom.xml file of your project:

<dependency>
    <groupId>com.starrocks</groupId>
    <artifactId>flink-connector-starrocks</artifactId>
    <!-- for flink-1.11, flink-1.12 -->
    <version>x.x.x_flink-1.11</version>
    <!-- for flink-1.13 -->
    <version>x.x.x_flink-1.13</version>
</dependency>

Sample code:

• Method 1

  // -------- sink with raw json string stream --------
  fromElements(new String[]{
      "{\"score\": \"99\", \"name\": \"stephen\"}",
      "{\"score\": \"100\", \"name\": \"lebron\"}"
  }).addSink(
      StarRocksSink.sink(
          // the sink options
          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 with stream transformation --------
  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(
          // the table structure
          TableSchema.builder()
              .field("score", DataTypes.INT())
              .field("name", DataTypes.VARCHAR(20))
              .build(),
          // the sink options
          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(),
          // set the slots with streamRowData
          (slots, streamRowData) -> {
              slots[0] = streamRowData.score;
              slots[1] = streamRowData.name;
          }
      )
  );

• Method 2

  // create a table with `structure` and `properties`
  // Needed: Add `com.starrocks.connector.flink.table.StarRocksDynamicTableSinkFactory` to: `src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory`
  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'" + // stream load properties like `'sink.properties.columns' = 'k1, v1'`
      ")"
  );

The following table describes the parameters of a Flink sink.

You can synchronize data from a MySQL database to a StarRocks cluster within seconds by using the Flink change data capture (CDC) connector and StarRocks migration tools.

When you run the sample code provided in this topic, take note of the following items:

• If you need to configure multiple sets of rules, you must configure the rules used to match the database names and table names, and configure the Flink StarRocks connector for each set of rules.

  Sample code:

  [table-rule.1]
  # pattern to match databases for setting properties
  database = ^console_19321.*$
  # pattern to match tables for setting properties
  table = ^.*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  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]
  # pattern to match databases for setting properties
  database = ^database2.*$
  # pattern to match tables for setting properties
  table = ^.*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  flink.starrocks.jdbc-url=jdbc:mysql://192.168.**.**:9030
  flink.starrocks.load-url= 192.168.**.**:8030
  flink.starrocks.username=root
  flink.starrocks.password=
  # If you cannot select appropriate delimiters to import data in the CSV format, you can import data in the JSON format. However, the import performance may deteriorate. To import data in the JSON format, replace the flink.starrocks.sink.properties.column_separator and flink.starrocks.sink.properties.row_delimiter parameters with the following parameters: 
  flink.starrocks.sink.properties.strip_outer_array=true
  flink.starrocks.sink.properties.format=json

  Note

  You can use the flink.starrocks.sink parameters to configure properties for each set of rules, such as the import frequency.

• After sharding is performed, data in a large table may be split into multiple tables or even distributed to multiple databases. In this case, you can configure a set of rules to synchronize data from multiple tables to one table.

  For example, both the edu_db_1 and edu_db_2 databases have two tables, course_1 and course_2. All the tables have the same schema. You can run the following code to synchronize data from these tables to a table in StarRocks for further analysis:

  [table-rule.3]
  # pattern to match databases for setting properties
  database = ^edu_db_[0-9]*$
  # pattern to match tables for setting properties
  table = ^course_[0-9]*$
  
  ############################################
  ### flink sink configurations
  ### DO NOT set `connector`, `table-name`, `database-name`, they are auto-generated
  ############################################
  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

  After you run the code, a many-to-one synchronization relationship is automatically generated. By default, the name of the table in StarRocks is course__auto_shard. You can modify the table name in the generated SQL files.

• If you need to create tables and synchronize data on the CLI of an SQL client, you must escape the backslash (\).

  Sample code:

  'sink.properties.column_separator' = '\\x01'
  'sink.properties.row_delimiter' = '\\x02'