將Flink寫入ClickHouse - ApsaraDB for ClickHouse

當您需要將Flink處理後的資料匯入ClickHouse時，本文為您提供了使用ClickHouse JDBC Connector的方法，使不同版本的Flink處理後的資料寫入到ClickHouse中。

背景資訊

Flink在1.11.0版本對其JDBC Connector進行了一次較大的重構：

重構之前（1.10.1及之前版本），包名為flink-jdbc 。
重構之後（1.11.0及之後版本），包名為flink-connector-jdbc 。

二者對Flink中以不同方式寫入ClickHouse Sink的支援情況如下：

API名稱	flink-jdbc	flink-connector-jdbc
DataStream	不支援	支援
Table API (Legacy)	支援	不支援
Table API (DDL)	不支援	不支援

flink-connector-jdbc完全移除了對Table API (Legacy) 的支援，只能通過DDL的方式調用Table API。但是，Table DDL方式寫入程式碼了其所支援的JDBC Driver，不支援ClickHouse。本文分別以基於Flink 1.10.1 + flink-jdbc和Flink 1.11.0 + flink-connector-jdbc 為例，介紹Flink寫入ClickHouse的方法。

Flink 1.10.1 + flink-jdbc

Flink 1.10.1及之前版本需要採用flink-jdbc+Table API的方式寫入資料到ClickHouse。本節我們使用Maven及Flink 1.10.1版本為例。

用mvn archetype:generate命令建立專案，產生過程中根據提示輸入group-id和artifact-id等。

$ mvn archetype:generate \
      -DarchetypeGroupId=org.apache.flink \
      -DarchetypeArtifactId=flink-quickstart-scala \
      -DarchetypeVersion=1.10.1

編輯pom.xml中的<dependencies />小節添加依賴。

        <!--//添加Flink Table API相關的依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>


        <!--//添加Flink JDBC以及Clickhouse JDBC Driver相關的依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>0.2.4</version>
        </dependency>

建立資料寫入程式檔案。

樣本程式使用CsvTableSource讀入 CSV 檔案產生Table Source，使用JDBCAppendTableSink將資料寫入到ClickHouse Sink中。

說明

由於ClickHouse單次插入的延遲比較高，我們需要設定BatchSize來批量插入資料，提高效能。
在JDBCAppendTableSink的實現中，若最後一批資料的數目不足BatchSize，則不會插入剩餘資料。

package org.myorg.example

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.sources._
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api._
import org.apache.flink.types.Row
import org.apache.flink.table.api.{
  TableEnvironment,
  TableSchema,
  Types,
  ValidationException
}
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink
import org.apache.flink.api.common.typeinfo.TypeInformation

object StreamingJob {
  def main(args: Array[String]) {
    val SourceCsvPath =
      "/<YOUR-PATH-TO-TEST-CSV>/source.csv"
    val CkJdbcUrl =
      "jdbc:clickhouse://<clickhouse-host>:<port>/<database>"
    val CkUsername = "<YOUR-USERNAME>"
    val CkPassword = "<YOUR-PASSWORD>"
    val BatchSize = 500 // 設定您的batch size

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val tEnv = StreamTableEnvironment.create(env)

    val csvTableSource = CsvTableSource
      .builder()
      .path(SourceCsvPath)
      .ignoreFirstLine()
      .fieldDelimiter(",")
      .field("name", Types.STRING)
      .field("age", Types.LONG)
      .field("sex", Types.STRING)
      .field("grade", Types.LONG)
      .field("rate", Types.FLOAT)
      .build()

    tEnv.registerTableSource("source", csvTableSource)

    val resultTable = tEnv.scan("source").select("name, grade, rate")

    val insertIntoCkSql =
      """
        |  INSERT INTO sink_table (
        |    name, grade, rate
        |  ) VALUES (
        |    ?, ?, ?
        |  )
      """.stripMargin

//將資料寫入 ClickHouse Sink 
    val sink = JDBCAppendTableSink
      .builder()
      .setDrivername("ru.yandex.clickhouse.ClickHouseDriver")
      .setDBUrl(CkJdbcUrl)
      .setUsername(CkUsername)
      .setPassword(CkPassword)
      .setQuery(insertIntoCkSql)
      .setBatchSize(BatchSize)
      .setParameterTypes(Types.STRING, Types.LONG, Types.FLOAT)
      .build()

    tEnv.registerTableSink(
      "sink",
      Array("name", "grade", "rate"),
      Array(Types.STRING, Types.LONG, Types.FLOAT),
      sink
    )

    tEnv.insertInto(resultTable, "sink")

    env.execute("Flink Table API to ClickHouse Example")
  }
}

參數說明：

SourceCsvPath：源CSV檔案路徑。
CkJdbcUrl：目標ClickHouse叢集地址。
CkUsername：目標ClickHouse叢集使用者名稱。
CkPassword：目標ClickHouse叢集對應密碼。

編譯運行。

$ mvn clean package
$ ${FLINK_HOME}/bin/flink run target/example-0.1.jar

Flink 1.11.0 + flink-connector-jdbc

Flink 1.11.0及之後版本需要採用flink-connector-jdbc+DataStream的方式寫入資料到ClickHouse。本節我們使用Maven及Flink 1.11.0版本為例。

用mvn archetype:generate命令建立專案，產生過程中會提示輸入group-id和artifact-id等。

$ mvn archetype:generate \
      -DarchetypeGroupId=org.apache.flink \
      -DarchetypeArtifactId=flink-quickstart-scala \
      -DarchetypeVersion=1.11.0

編輯pom.xml中的<dependencies />小節添加依賴。

        <!--//添加Flink Table API相關的依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <!--//添加Flink JDBC Connector以及Clickhouse JDBC Driver相關的依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>0.2.4</version>
        </dependency>

建立資料寫入程式檔案。

樣本程式使用 CsvTableSource讀入CSV檔案產生Table Source，通過TableEnvironment.toAppendStream 將Table轉換為DataStream。使用JdbcSink將資料寫入到ClickHouse中。

說明

由於ClickHouse單次插入的延遲比較高，我們需要設定BatchSize來批量插入資料，提高效能。
目前的版本的flink-connector-jdbc，使用Scala API調用JdbcSink時會出現lambda函數的序列化問題。我們只能採用手動實現interface的方式來傳入相關JDBC Statement build函數（class CkSinkBuilder）。
```
class CkSinkBuilder extends JdbcStatementBuilder[(String, Long, Float)] {
  def accept(ps: PreparedStatement, v: (String, Long, Float)): Unit = {
    ps.setString(1, v._1)
    ps.setLong(2, v._2)
    ps.setFloat(3, v._3)
  }
}
```

package org.myorg.example

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.sources._
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api._
import org.apache.flink.types.Row
import org.apache.flink.table.api.{
  TableEnvironment,
  TableSchema,
  Types,
  ValidationException
}
import org.apache.flink.connector.jdbc._
import java.sql.PreparedStatement

//手動實現interface的方式來傳入相關JDBC Statement build函數
class CkSinkBuilder extends JdbcStatementBuilder[(String, Long, Float)] {
  def accept(ps: PreparedStatement, v: (String, Long, Float)): Unit = {
    ps.setString(1, v._1)
    ps.setLong(2, v._2)
    ps.setFloat(3, v._3)
  }
}

object StreamingJob {
  def main(args: Array[String]) {
    val SourceCsvPath =
      "/<YOUR-PATH-TO-TEST-CSV>/source.csv"
    val CkJdbcUrl = "jdbc:clickhouse://<clickhouse-host>:<port>/<database>"
    val CkUsername = "<YOUR-USERNAME>"
    val CkPassword = "<YOUR-PASSWORD>"
    val BatchSize = 500 // 設定您的 batch size

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val tEnv = StreamTableEnvironment.create(env)

    val csvTableSource = CsvTableSource
      .builder()
      .path(SourceCsvPath)
      .ignoreFirstLine()
      .fieldDelimiter(",")
      .field("name", Types.STRING)
      .field("age", Types.LONG)
      .field("sex", Types.STRING)
      .field("grade", Types.LONG)
      .field("rate", Types.FLOAT)
      .build()

    tEnv.registerTableSource("source", csvTableSource)

    val resultTable = tEnv.scan("source").select("name, grade, rate")

//將Table轉換為DataStream
    val resultDataStream =
      tEnv.toAppendStream[(String, Long, Float)](resultTable)

    val insertIntoCkSql =
      """
        |  INSERT INTO sink_table (
        |    name, grade, rate
        |  ) VALUES (
        |    ?, ?, ?
        |  )
      """.stripMargin

//將資料寫入ClickHouse JDBC Sink
    resultDataStream.addSink(
      JdbcSink.sink[(String, Long, Float)](
        insertIntoCkSql,
        new CkSinkBuilder,
        new JdbcExecutionOptions.Builder().withBatchSize(BatchSize).build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
          .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
          .withUrl(CkJdbcUrl)
          .withUsername(CkUsername)
          .withPassword(CkPassword)
          .build()
      )
    )

    env.execute("Flink DataStream to ClickHouse Example")
  }
}

參數說明：

SourceCsvPath：源CSV檔案路徑。
CkJdbcUrl：目標ClickHouse叢集地址。
CkUsername：目標ClickHouse叢集使用者名稱。
CkPassword：目標ClickHouse叢集對應密碼。

編譯運行。

$ mvn clean package
$ ${FLINK_HOME}/bin/flink run target/example-0.1.jar