​Spark 写入 ClickHouse API

SparkCore写入ClickHouse，可以直接采用写入方式。下面案例是使用SparkSQL将结果存入ClickHouse对应的表中。在ClickHouse中需要预先创建好对应的结果表。

一、导入依赖

<!-- 连接ClickHouse需要驱动包-->
<dependency>
  <groupId>ru.yandex.clickhouse</groupId>
  <artifactId>clickhouse-jdbc</artifactId>
  <version>0.2.4</version>
  <!-- 去除与Spark 冲突的包 -->
  <exclusions>
    <exclusion>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </exclusion>
    <exclusion>
        <groupId>net.jpountz.lz4</groupId>
        <artifactId>lz4</artifactId>
    </exclusion>
</exclusions>
</dependency>

<!-- Spark-core -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.3.1</version>
</dependency>
<!-- SparkSQL -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.3.1</version>
</dependency>
<!-- SparkSQL  ON  Hive-->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-hive_2.11</artifactId>
  <version>2.3.1</version>
</dependency>

​

二、代码编写

val session: SparkSession = SparkSession.builder().master("local").appName("test").getOrCreate()
val jsonList = List[String](
  "{\"id\":1,\"name\":\"张三\",\"age\":18}",
  "{\"id\":2,\"name\":\"李四\",\"age\":19}",
  "{\"id\":3,\"name\":\"王五\",\"age\":20}"
)

//将jsonList数据转换成DataSet
import session.implicits._
val ds: Dataset[String] = jsonList.toDS()

val df: DataFrame = session.read.json(ds)
df.show()

//将结果写往ClickHouse
val url = "jdbc:clickhouse://node1:8123/default"
val table = "test"
val properties = new Properties()
properties.put("driver", "ru.yandex.clickhouse.ClickHouseDriver")
properties.put("user", "default")
properties.put("password", "")
properties.put("socket_timeout", "300000")
df.write.mode(SaveMode.Append).option(JDBCOptions.JDBC_BATCH_INSERT_SIZE, 100000).jdbc(url, table, properties)

​

​

​