Flink Connector內部實現是通過緩存并批量由Stream Load導入。本文為您介紹Flink Connector的使用方式及示例。

背景信息

因為Flink官方只提供了flink-connector-jdbc，不足以滿足導入性能要求，所以新增了flink-connector-starrocks，其內部實現是通過緩存并批量由Stream Load導入。

使用方式

您可以下載源碼進行測試：下載flink-connector-starrocks源碼。

將以下內容添加pom.xml中。
<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>
說明 您可以在版本信息頁面，查看Latest Version信息，替換代碼中的x.x.x。
代碼示例如下：
  • 方式一
    // -------- 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")
            .withProperty("load-url", "fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port")
            .withProperty("username", "admin")
            .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")
            .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;
        }
    )
);
  • 方式二
    // 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'," +
        "'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'`
    ")"
);
其中，Sink選項如下表所示。
參數是否必選默認值類型描述
connector是無String類型，固定為starrocks。
jdbc-url是無String用于在StarRocks中執行查詢操作。
load-url是無String指定FE的IP和HTTP端口，格式為fe_ip:http_port;fe_ip:http_port ，多個時使用半角分號（;）分隔。
database-name是無StringStarRocks數據庫名稱。
table-name是無StringStarRocks表名稱。
username是無StringStarRocks連接用戶名。
password是無StringStarRocks連接密碼。
sink.semantic否at-least-onceString取值為at-least-once或exactly-once。
sink.buffer-flush.max-bytes否94371840（90M）StringBuffer可容納的最大數據量，取值范圍為64 MB~10 GB。
sink.buffer-flush.max-rows否500000StringBuffer可容納的最大數據行數，取值范圍為64,000~5000,000。
sink.buffer-flush.interval-ms否300000StringBuffer刷新時間間隔，取值范圍為 1000 ms~3600000 ms。
sink.max-retries否1String最大重試次數，取值范圍為0~10。
sink.connect.timeout-ms否1000String連接到load-url的超時時間，取值范圍為100~60000。
sink.properties.*否無StringSink屬性。
重要
  • 為了保證Sink數據的Exactly-once語義，需要外部系統的Two-phase Commit機制。由于StarRocks無此機制，所以需要依賴Flink的checkpoint-interval，在每次Checkpoint時保存批數據以及其Label，在Checkpoint完成后的第一次invoke中阻塞flush所有緩存在state中的數據，以此達到Exactly-once。但如果StarRocks終止了，會導致您的Flink Sink Stream算子長時間阻塞，并引起Flink的監控報警或強制退出。
  • 默認使用CSV格式進行導入，您可以通過指定sink.properties.row_delimiter（該參數自StarRocks 1.15.0版本開始支持）為\\x02，sink.properties.column_separator為\\x01，來自定義行分隔符與列分隔符。
  • 如果遇到導入停止的情況，請嘗試增加Flink任務的內存。
  • 如果代碼運行正常且能接收到數據，但是寫入不成功時，請確認當前機器是否能訪問BE的http_port端口，即能ping通BE服務使用的IP地址。

    例如，如果一臺機器有外網和內網IP，且FE或BE的http_port均可通過外網ip:port訪問，集群里綁定的IP為內網IP，任務里loadurl寫的FE外網ip:http_port，FE會將寫入任務轉發給BE內網ip:port，此時如果Client機器ping不通BE的內網IP就會寫入失敗。

示例

詳情請參見使用Flink CDC同步MySQL數據至StarRocks。