Importación desde Spark - ApsaraDB for ClickHouse - Centro de documentación de Alibaba Cloud

Utilice un programa de Spark para importar datos CSV por lotes a ApsaraDB for ClickHouse mediante una conexión JDBC. Este enfoque es adecuado cuando ya cuenta con un pipeline de Spark y desea escribir DataFrames directamente en ClickHouse con el controlador JDBC de ClickHouse.

Si requiere compatibilidad completa con tipos complejos (MAP, ARRAY, STRUCT) o prefiere una integración más nativa, utilice el conector nativo de Spark-ClickHouse . El método JDBC que se describe aquí no admite dichos tipos.

Requisitos previos

Antes de empezar, asegúrese de haber:

Agregado la dirección IP de su computadora local a la lista de permitidos del clúster de ApsaraDB for ClickHouse. Consulte Configure a whitelist.
Creado una tabla en ApsaraDB for ClickHouse cuyos tipos de datos de columna coincidan con los datos que desea importar. Consulte Create a table.

Funcionamiento

El programa de Spark lee un archivo CSV en un DataFrame y lo escribe en una tabla de ApsaraDB for ClickHouse mediante el controlador JDBC de ClickHouse. Las filas se insertan por lotes a través de una conexión JDBC en el puerto 8123.

Parámetros clave de la conexión:

Parámetro	Valor	Descripción
`batchsize`	`100000`	Número de filas por inserción por lotes
`socket_timeout`	`300000`	Tiempo de espera del socket en milisegundos
`numPartitions`	`8`	Número de particiones de escritura paralelas. Auméntelo para conjuntos de datos grandes; redúzcalo para disminuir la carga en el clúster.
`rewriteBatchedStatements`	`true`	Reescribe las inserciones por lotes en una sola instrucción de múltiples filas para mejorar el throughput

Importar datos desde CSV

Paso 1: Configurar la estructura del proyecto

Cree la siguiente estructura de directorios para su proyecto de 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

Paso 2: Agregar dependencias

Agregue el siguiente contenido a 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"

Paso 3: Escribir el programa de Spark

Cree el archivo src/main/scala/com/spark/test/WriteToCk.scala con el siguiente contenido. Reemplace los marcadores de posición que se enumeran en la tabla siguiente antes de ejecutarlo.

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)
  }
}

Reemplace los siguientes marcadores de posición con sus valores reales:

Marcador de posición	Descripción	Obligatorio
`<yourUserName>`	Nombre de usuario de la cuenta de base de datos en ApsaraDB for ClickHouse	Sí
`<yourPassword>`	Contraseña de la cuenta de base de datos	Sí
`<yourUrl>`	Endpoint del clúster de ApsaraDB for ClickHouse	Sí
`<yourTableName>`	Nombre de la tabla de destino en ApsaraDB for ClickHouse	Sí
`<yourFilePath>`	Ruta al archivo CSV que se va a importar, incluido el nombre del archivo	Sí
`colName1,colName2,colName3`	Nombres de columnas en la tabla de ApsaraDB for ClickHouse que se seleccionarán del DataFrame	Sí

Paso 4: Compilar el paquete

Ejecute el siguiente comando para compilar y empaquetar el programa:

sbt package

El archivo JAR de salida se genera en target/scala-2.12/simple-project_2.12-1.0.jar.

Paso 5: Enviar el trabajo de Spark

Ejecute el siguiente comando para enviar el trabajo. Este comando agrega el controlador JDBC de ClickHouse a la ruta de clases tanto para los procesos del controlador como del ejecutor.

${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

Limitaciones

Sin compatibilidad con tipos de datos complejos: El controlador JDBC de ClickHouse no admite tipos complejos de Spark como MAP, ARRAY o STRUCT. Para utilizar estos tipos, cambie al conector nativo de Spark-ClickHouse.
Las tablas deben existir antes de la importación: JDBC no puede crear automáticamente la tabla de destino. Cree la tabla en ApsaraDB for ClickHouse antes de ejecutar el trabajo de Spark. Consulte Create a table.