E-MapReduce:Flink から ClickHouse へのデータインポート

以下のセクションでは、ストリーム処理とバッチ処理のコード例を示します。

• ストリーム処理

  package com.company.packageName
  
  import java.util.concurrent.ThreadLocalRandom
  
  import scala.annotation.tailrec
  
  import org.apache.flink.api.common.typeinfo.Types
  import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink
  import org.apache.flink.streaming.api.scala._
  import org.apache.flink.table.api.scala.{StreamTableEnvironment, table2RowDataStream}
  
  object StreamingJob {
  
    case class Test(id: Int, key1: String, value1: Boolean, key2: Long, value2: Double)
  
    private var dbName: String = "default"
    private var tableName: String = ""
    private var ckHost: String = ""
    private var ckPort: String = "8123"
    private var user: String = "default"
    private var password: String = ""
  
    def main(args: Array[String]) {
      parse(args.toList)
      checkArguments()
  
      // ストリーミング実行環境をセットアップします
      val env = StreamExecutionEnvironment.getExecutionEnvironment
      val tableEnv = StreamTableEnvironment.create(env)
  
      val insertIntoCkSql =
        s"""
          | INSERT INTO $tableName (
          |   id, key1, value1, key2, value2
          | ) VALUES (
          |   ?, ?, ?, ?, ?
          | )
          |""".stripMargin
  
      val jdbcUrl = s"jdbc:clickhouse://$ckHost:$ckPort/$dbName"
  
      println(s"jdbc url: $jdbcUrl")
      println(s"insert sql: $insertIntoCkSql")
  
      val sink = JDBCAppendTableSink
        .builder()
        .setDrivername("ru.yandex.clickhouse.ClickHouseDriver")
        .setDBUrl(jdbcUrl)
        .setUsername(user)
        .setPassword(password)
        .setQuery(insertIntoCkSql)
        .setBatchSize(1000)
        .setParameterTypes(Types.INT, Types.STRING, Types.BOOLEAN, Types.LONG, Types.DOUBLE)
        .build()
  
      val data: DataStream[Test] = env.fromCollection(1 to 1000).map(i => {
        val rand = ThreadLocalRandom.current()
        val randString = (0 until rand.nextInt(10, 20))
          .map(_ => rand.nextLong())
          .mkString("")
        Test(i, randString, rand.nextBoolean(), rand.nextLong(), rand.nextGaussian())
      })
  
      val table = table2RowDataStream(tableEnv.fromDataStream(data))
      sink.emitDataStream(table.javaStream)
  
      // プログラムを実行します
      env.execute("Flink Streaming Scala API Skeleton")
    }
  
    private def printUsageAndExit(exitCode: Int = 0): Unit = {
      println("Usage: flink run com.company.packageName.StreamingJob /path/to/flink-clickhouse-demo-1.0.0.jar [options]")
      println("  --dbName      ClickHouse データベースの名前を設定します。デフォルト: default。")
      println("  --tableName   ClickHouse データベース内のテーブルの名前を設定します。")
      println("  --ckHost      ClickHouse クラスターの IP アドレスを設定します。")
      println("  --ckPort      ClickHouse クラスターのポートを設定します。デフォルト: 8123。")
      println("  --user        ClickHouse クラスターにアクセスするためのユーザー名を設定します。")
      println("  --password    ClickHouse ユーザーのパスワードを設定します。")
      System.exit(exitCode)
    }
  
    @tailrec
    private def parse(args: List[String]): Unit = args match {
      case ("--help" | "-h") :: _ =>
        printUsageAndExit()
      case "--dbName" :: value :: tail =>
        dbName = value
        parse(tail)
      case "--tableName" :: value :: tail =>
        tableName = value
        parse(tail)
      case "--ckHost" :: value :: tail =>
        ckHost = value
        parse(tail)
      case "--ckPort" :: value :: tail =>
        ckPort = value
        parse(tail)
      case "--user" :: value :: tail =>
        user = value
        parse(tail)
      case "--password" :: value :: tail =>
        password = value
        parse(tail)
      case Nil =>
      case _ =>
        printUsageAndExit(1)
    }
  
    private def checkArguments(): Unit = {
      if ("".equals(tableName) || "".equals(ckHost)) {
        printUsageAndExit(2)
      }
    }
  }

• バッチ処理

  package com.company.packageName
  
  import java.util.concurrent.ThreadLocalRandom
  
  import scala.annotation.tailrec
  
  import org.apache.flink.Utils
  import org.apache.flink.api.common.typeinfo.Types
  import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink
  import org.apache.flink.api.scala._
  import org.apache.flink.table.api.scala.{BatchTableEnvironment, table2RowDataSet}
  
  object BatchJob {
  
    case class Test(id: Int, key1: String, value1: Boolean, key2: Long, value2: Double)
  
    private var dbName: String = "default"
    private var tableName: String = ""
    private var ckHost: String = ""
    private var ckPort: String = "8123"
    private var user: String = "default"
    private var password: String = ""
  
    def main(args: Array[String]) {
      parse(args.toList)
      checkArguments()
  
      // バッチ実行環境をセットアップします
      val env = ExecutionEnvironment.getExecutionEnvironment
      val tableEnv = BatchTableEnvironment.create(env)
  
      val insertIntoCkSql =
        s"""
          | INSERT INTO $tableName (
          |   id, key1, value1, key2, value2
          | ) VALUES (
          |   ?, ?, ?, ?, ?
          | )
          |""".stripMargin
      val jdbcUrl = s"jdbc:clickhouse://$ckHost:$ckPort/$dbName"
  
      println(s"jdbc url: $jdbcUrl")
      println(s"insert sql: $insertIntoCkSql")
  
      val sink = JDBCAppendTableSink
        .builder()
        .setDrivername("ru.yandex.clickhouse.ClickHouseDriver")
        .setDBUrl(jdbcUrl)
        .setUsername(user)
        .setPassword(password)
        .setQuery(insertIntoCkSql)
        .setBatchSize(1000)
        .setParameterTypes(Types.INT, Types.STRING, Types.BOOLEAN, Types.LONG, Types.DOUBLE)
        .build()
  
      val data = env.fromCollection(1 to 1000).map(i => {
        val rand = ThreadLocalRandom.current()
        val randString = (0 until rand.nextInt(10, 20))
          .map(_ => rand.nextLong())
          .mkString("")
        Test(i, randString, rand.nextBoolean(), rand.nextLong(), rand.nextGaussian())
      })
  
      val table = table2RowDataSet(tableEnv.fromDataSet(data))
  
      sink.emitDataSet(Utils.convertScalaDatasetToJavaDataset(table))
  
      // プログラムを実行します
      env.execute("Flink Batch Scala API Skeleton")
    }
  
    private def printUsageAndExit(exitCode: Int = 0): Unit = {
      println("Usage: flink run com.company.packageName.BatchJob /path/to/flink-clickhouse-demo-1.0.0.jar [options]")
      println("  --dbName      ClickHouse データベースの名前を設定します。デフォルト: default。")
      println("  --tableName   ClickHouse データベース内のテーブルの名前を設定します。")
      println("  --ckHost      ClickHouse クラスターの IP アドレスを設定します。")
      println("  --ckPort      ClickHouse クラスターのポートを設定します。デフォルト: 8123。")
      println("  --user        ClickHouse クラスターにアクセスするためのユーザー名を設定します。")
      println("  --password    ClickHouse ユーザーのパスワードを設定します。")
      System.exit(exitCode)
    }
  
    @tailrec
    private def parse(args: List[String]): Unit = args match {
      case ("--help" | "-h") :: _ =>
        printUsageAndExit()
      case "--dbName" :: value :: tail =>
        dbName = value
        parse(tail)
      case "--tableName" :: value :: tail =>
        tableName = value
        parse(tail)
      case "--ckHost" :: value :: tail =>
        ckHost = value
        parse(tail)
      case "--ckPort" :: value :: tail =>
        ckPort = value
        parse(tail)
      case "--user" :: value :: tail =>
        user = value
        parse(tail)
      case "--password" :: value :: tail =>
        password = value
        parse(tail)
      case Nil =>
      case _ =>
        printUsageAndExit(1)
    }
  
    private def checkArguments(): Unit = {
      if ("".equals(tableName) || "".equals(ckHost)) {
        printUsageAndExit(2)
      }
    }
  }