ClickHouse-Spark集成

ClickHouse-Spark集成 Spark集成ClickHouse

ClickHouse的JDBC驱动

目前通过JDBC写ClickHouse一共有三种驱动，除了官方的，还有两种第三方驱动。分别如下：

1. 官方的JDBC Driver：8123端口

   基于HTTP实现，整体性能不太出色，大量数据写入时有可能出现超时的现象

2. housepower的ClickHouse-Native-JDBC：9000端口

   基于TCP协议实现，支持高性能写入，数据按列组织并有压缩

3. Clickhouse4j

   基于HTTP，但是相比官方进行了大量优化，更加轻量级和快速

因为支持高性能的写入和压缩，因此本次实践使用第二种驱动

依赖如下：

    com.github.housepower
    clickhouse-native-jdbc
    2.6.0

已有数据如下：

SELECt *
FROM user

┌─userId─┬─appId─┬─version─┬────regTime─┐
│   1234 │ TT    │ 2.4     │ 2020-07-24 │
│  12345 │ RR    │ 2.5     │ 2020-07-29 │
│ 123456 │ TT    │ 2.1     │ 2020-07-09 │
│ 234561 │ GG    │ 3.0     │ 2020-07-31 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│ 123457 │ RR    │ 3.6     │ 2020-01-07 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│  43234 │ HH    │ 2.5     │ 2020-06-06 │
└────────┴───────┴─────────┴────────────┘

代码如下：

import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.{SparkConf}

object SparkCoreAndClickHouse {

    def main(args: Array[String]): Unit = {

        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("ClickHouse")
        val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

        val url = "jdbc:clickhouse://10.10.10.113:9000"
        val drivce = "com.github.housepower.jdbc.ClickHouseDriver"

        // 官方的jar包使用8123作为端口
        val df = spark.read
            .format("jdbc")
            .option("driver", drivce)
            .option("url", url)
            .option("user", "default")
            .option("password", "")
            .option("dbtable", "default.user")
            .load

        df.show()

        val pro = new java.util.Properties
        pro.put("driver", drivce)

        df.write
            .mode(SaveMode.Append)
            .option("batchsize", "20000")
            .option("isolationLevel", "NONE")
            .option("numPartitions", "1")
            .jdbc(url, "default.user", pro)

        // 关闭环境
        spark.close()
    }
}

代码是从user表中查出数据，然后打印到控制台，最后将查出的数据再次插入到user表中。

执行完成只有，查询user表结果如下：

SELECt *
FROM user

┌─userId─┬─appId─┬─version─┬────regTime─┐
│   1234 │ TT    │ 2.4     │ 2020-07-24 │
│  12345 │ RR    │ 2.5     │ 2020-07-29 │
│ 123456 │ TT    │ 2.1     │ 2020-07-09 │
│ 234561 │ GG    │ 3.0     │ 2020-07-31 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│ 123457 │ RR    │ 3.6     │ 2020-01-07 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│   1234 │ TT    │ 2.4     │ 2020-07-24 │
│  12345 │ RR    │ 2.5     │ 2020-07-29 │
│ 123456 │ TT    │ 2.1     │ 2020-07-09 │
│ 234561 │ GG    │ 3.0     │ 2020-07-31 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│ 123457 │ RR    │ 3.6     │ 2020-01-07 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│  43234 │ HH    │ 2.5     │ 2020-06-06 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
│    123 │ test  │ 1.1.1.1 │ 2021-12-10 │
└────────┴───────┴─────────┴────────────┘
┌─userId─┬─appId─┬─version─┬────regTime─┐
│  43234 │ HH    │ 2.5     │ 2020-06-06 │
└────────┴───────┴─────────┴────────────┘