全部產品
Search
文件中心

ApsaraDB for ClickHouse:從Spark匯入

更新時間:Jun 30, 2024

本文介紹如何通過Spark程式匯入資料至雲資料庫ClickHouse

前提條件

  • 已將本地機器的IP地址添加到雲資料庫ClickHouse的白名單中。如何添加,請參見設定白名單
  • 已建立與匯入資料結構類型對應的雲資料庫ClickHouse表。如何建立,請參見建立表

操作步驟

  1. 準備Spark程式目錄結構。
     find .
    .
    ./build.sbt
    ./src
    ./src/main
    ./src/main/scala
    ./src/main/scala/com
    ./src/main/scala/com/spark
    ./src/main/scala/com/spark/test
    ./src/main/scala/com/spark/test/WriteToCk.scala
  2. 編輯build.sbt設定檔並添加依賴。
    name := "Simple Project"
    
    version := "1.0"
    
    scalaVersion := "2.12.10"
    
    libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.0.0"
    
    libraryDependencies += "ru.yandex.clickhouse" % "clickhouse-jdbc" % "0.2.4"
  3. 建立WriteToCk.scala資料並寫入程式檔案。
    package com.spark.test
    
    import java.util
    import java.util.Properties
    
    import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.{SaveMode, SparkSession}
    import org.apache.spark.storage.StorageLevel
    
    object WriteToCk {
      val properties = new Properties()
      properties.put("driver", "ru.yandex.clickhouse.ClickHouseDriver")
      properties.put("user", "<yourUserName>")
      properties.put("password", "<yourPassword>")
      properties.put("batchsize","100000")
      properties.put("socket_timeout","300000")
      properties.put("numPartitions","8")
      properties.put("rewriteBatchedStatements","true")
    
      val url = "jdbc:clickhouse://<yourUrl>:8123/default"
      val table = "<yourTableName>"
    
      def main(args: Array[String]): Unit = {
        val sc = new SparkConf()
        sc.set("spark.driver.memory", "1G")
        sc.set("spark.driver.cores", "4")
        sc.set("spark.executor.memory", "1G")
        sc.set("spark.executor.cores", "2")
    
        val session = SparkSession.builder().master("local[*]").config(sc).appName("write-to-ck").getOrCreate()
    
        val df = session.read.format("csv")
          .option("header", "true")
          .option("sep", ",")
          .option("inferSchema", "true")
          .load("<yourFilePath>")
          .selectExpr(
            "colName1",
            "colName2",
            "colName3",
             ...
          )
          .persist(StorageLevel.MEMORY_ONLY_SER_2)
        println(s"read done")
    
        df.write.mode(SaveMode.Append).option(JDBCOptions.JDBC_BATCH_INSERT_SIZE, 100000).jdbc(url, table, properties)
        println(s"write done")
    
        df.unpersist(true)
      }
    }
    參數說明如下。
    參數說明
    yourUserName雲資料庫ClickHouse中建立的資料庫帳號。
    yourPassword資料庫帳號的密碼。
    yourUrl資料庫連接地址。
    yourTableName雲資料庫ClickHouse中建立的表名。
    yourFilePath匯入資料檔案的儲存路徑,包含檔案地址和檔案名稱。
    colName1,colName2,colName3雲資料庫ClickHouse表的列名。
  4. 編譯打包。
    sbt package
  5. 運行。
    ${SPARK_HOME}/bin/spark-submit  --class "com.spark.test.WriteToCk"  --master local[4] --conf "spark.driver.extraClassPath=${HOME}/.m2/repository/ru/yandex/clickhouse/clickhouse-jdbc/0.2.4/clickhouse-jdbc-0.2.4.jar" --conf "spark.executor.extraClassPath=${HOME}/.m2/repository/ru/yandex/clickhouse/clickhouse-jdbc/0.2.4/clickhouse-jdbc-0.2.4.jar" target/scala-2.12/simple-project_2.12-1.0.jar