大数据ClickHouse(十八):Spark 写入 ClickHouse API

2022-09-03 00:23:59 浏览数 (3)

Spark 写入 ClickHouse API

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

一、导入依赖

代码语言:javascript复制
<!-- 连接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>

二、代码编写

代码语言:javascript复制
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)

0 人点赞