从Flink导入数据至ClickHouse

更新时间: 2023-08-21 16:26:15

本文为您介绍如何将Flink中的数据导入至ClickHouse集群。

前提条件

背景信息

关于Flink的更多介绍,请参见Apache Flink

代码示例

代码示例如下:

  • 流处理

    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()
    
        // set up the streaming execution environment
        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)
    
        // execute program
        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地址")
        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()
    
        // set up the batch execution environment
        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))
    
        // execute program
        env.execute("Flink Batch 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地址")
        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)
        }
      }
    }

操作流程

  1. 步骤一:创建ClickHouse表

  2. 步骤二:编译并打包

  3. 步骤三:提交作业

步骤一:创建ClickHouse表

  1. 使用SSH方式登录ClickHouse集群,详情请参见登录集群

  2. 执行如下命令,启动ClickHouse客户端。

    clickhouse-client -h core-1-1 -m
    说明

    本示例登录core-1-1节点,如果您有多个Core节点,可以登录任意一个节点。

  3. 创建ClickHouse信息。

    1. 执行如下命令,创建数据库clickhouse_database_name

      CREATE DATABASE clickhouse_database_name ON CLUSTER cluster_emr;

      阿里云EMR会为ClickHouse集群自动生成一个名为cluster_emr的集群。数据库名您可以自定义。

    2. 执行如下命令,创建表clickhouse_table_name_local

      CREATE TABLE clickhouse_database_name.clickhouse_table_name_local ON CLUSTER cluster_emr (
        id            UInt32,
        key1            String,
        value1        UInt8,
        key2            Int64,
        value2        Float64
      ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/clickhouse_database_name/clickhouse_table_name_local', '{replica}')
      ORDER BY id;
      说明

      表名您可以自定义,但请确保表名是以_local结尾。layershardreplica是阿里云EMR为ClickHouse集群自动生成的宏定义,可以直接使用。

    3. 执行如下命令,创建与表clickhouse_table_name_local字段定义一致的表clickhouse_table_name_all

      说明

      表名您可以自定义,但请确保表名是以_all结尾。

      CREATE TABLE clickhouse_database_name.clickhouse_table_name_all ON CLUSTER cluster_emr (
        id                    UInt32,
        key1                  String,
        value1                UInt8,
        key2                  Int64,
        value2                Float64
      ) ENGINE = Distributed(cluster_emr, clickhouse_database_name, clickhouse_table_name_local, rand());

步骤二:编译并打包

  1. 下载并解压flink-clickhouse-demo.tgz示例到本地。

  2. 在CMD命令行中,进入到下载文件中pom.xml所在的目录下,执行如下命令打包文件。

    mvn clean package

    根据您pom.xml文件中artifactId的信息,下载文件中的target目录下会出现flink-clickhouse-demo-1.0.0.jar的JAR包。

步骤三:提交作业

  1. 使用SSH方式登录Flink集群,详情请参见登录集群

  2. 上传打包好的flink-clickhouse-demo-1.0.0.jar至Flink集群的根目录下。

    说明

    本文示例中flink-clickhouse-demo-1.0.0.jar是上传至root根目录下,您也可以自定义上传路径。

  3. 执行如下命令提交作业。

    代码示例如下:

    • 流作业

      flink run -m yarn-cluster \
                -c com.aliyun.emr.StreamingJob \
                flink-clickhouse-demo-1.0.0.jar \
                --dbName clickhouse_database_name \
                --tableName clickhouse_table_name_all \
                --ckHost ${clickhouse_host} \
                --password ${password};
    • 批作业

      flink run -m yarn-cluster \
                -c com.aliyun.emr.BatchJob \
                flink-clickhouse-demo-1.0.0.jar \
                --dbName clickhouse_database_name \
                --tableName clickhouse_table_name_all \
                --ckHost ${clickhouse_host} \
                --password ${password};

    参数

    说明

    dbName

    ClickHouse集群数据库的名称,默认为default。本文示例为clickhouse_database_name

    tableName

    ClickHouse集群数据库中表的名称。本文示例为clickhouse_table_name_all

    ckHost

    ClickHouse集群的Master节点的内网IP地址或公网IP地址。ip地址获取方式,请参见获取主节点的IP地址

    password

    ClickHouse用户的密码。

    您可以在ClickHouse服务的配置页面,通过查看users.default.password参数,获取密码。

    password

获取主节点的IP地址

  1. 进入节点管理页面。

    1. 登录EMR on ECS

    2. 在顶部菜单栏处,根据实际情况选择地域和资源组

    3. EMR on ECS页面,单击目标集群操作列的节点管理

  2. 节点管理页面,单击Master节点组所在行的open图标,复制公网IP列的IP地址。

上一篇: 从Spark导入数据至ClickHouse 下一篇: HDFS与ClickHouse间的数据导入与导出
阿里云首页 开源大数据平台 E-MapReduce 相关技术圈