如何通过JDBC将含结构体的Spark DataFrame写入ClickHouse

方法1：使用ClickHouse官方Spark Connector（推荐） #

JDBC原生对Spark复杂类型（结构体、结构体数组）支持有限，ClickHouse官方Spark Connector可直接映射Spark的StructType、Array[StructType]到ClickHouse的Nested类型，无需手动转换。

操作步骤 #

1. 添加依赖
   Maven项目引入依赖：

   <dependency>
       <groupId>com.clickhouse</groupId>
       <artifactId>clickhouse-spark-connector_2.12</artifactId>
       <version>0.5.0</version> <!-- 替换为最新稳定版 -->
   </dependency>
   

   Spark Submit命令引入：

   spark-submit --packages com.clickhouse:clickhouse-spark-connector_2.12:0.5.0 your-application.jar
   

2. 创建对应ClickHouse表
   针对示例中的复杂字段，ClickHouse表需用Nested类型定义，完整建表示例：

   CREATE TABLE IF NOT EXISTS your_database.your_table (
       guest_third_party_id_arr Nested(source String, id String),
       foryou Nested(refreshclick Boolean, removed Boolean),
       loved Boolean,
       loveditems Nested(brandname String, removed Boolean, undo Boolean),
       popular Nested(brandname String, featuredbrand Boolean, mostloved Boolean, topviewed Boolean)
   ) ENGINE = MergeTree()
   ORDER BY tuple();
   

3. Spark写入代码示例
   Scala版本：

   import org.apache.spark.sql.SparkSession
   
   val spark = SparkSession.builder()
       .appName("WriteComplexDFToClickHouse")
       .getOrCreate()
   
   // 假设df为已构建的含复杂结构的DataFrame
   df.write
       .format("clickhouse")
       .option("url", "jdbc:clickhouse://your-clickhouse-host:8123/your_database")
       .option("user", "your-username")
       .option("password", "your-password")
       .option("dbtable", "your_table")
       .mode("append")
       .save()
   

   Python版本：

   from pyspark.sql import SparkSession
   
   spark = SparkSession.builder \
       .appName("WriteComplexDFToClickHouse") \
       .getOrCreate()
   
   # 假设df为已构建的含复杂结构的DataFrame
   df.write \
       .format("clickhouse") \
       .option("url", "jdbc:clickhouse://your-clickhouse-host:8123/your_database") \
       .option("user", "your-username") \
       .option("password", "your-password") \
       .option("dbtable", "your_table") \
       .mode("append") \
       .save()
   

方法2：自定义JDBC Dialect（强制使用JDBC时） #

若必须基于JDBC实现，需扩展ClickHouseDialect，自定义复杂类型到ClickHouse Nested类型的映射逻辑：

操作步骤 #

1. 自定义Dialect类
   Scala代码示例：

   import org.apache.spark.sql.jdbc.JdbcDialect
   import org.apache.spark.sql.types._
   
   class CustomClickHouseDialect extends JdbcDialect {
       override def canHandle(url: String): Boolean = url.startsWith("jdbc:clickhouse:")
   
       override def getJdbcType(dt: DataType): Option[(String, Int)] = dt match {
           // 处理单层结构体
           case struct: StructType =>
               val nestedFields = struct.fields.map(f => s"${f.name} ${getJdbcType(f.dataType).get._1}").mkString(", ")
               Some((s"Nested($nestedFields)", java.sql.Types.OTHER))
           // 处理结构体数组
           case ArrayType(elementType: StructType, _) =>
               val nestedFields = elementType.fields.map(f => s"${f.name} ${getJdbcType(f.dataType).get._1}").mkString(", ")
               Some((s"Array(Nested($nestedFields))", java.sql.Types.OTHER))
           // 其他类型沿用原有逻辑
           case _ => ClickHouseDialect.getJdbcType(dt)
       }
   }
   

2. 注册自定义Dialect
   在Spark应用启动时注册：

   import org.apache.spark.sql.jdbc.JdbcDialects
   
   JdbcDialects.registerDialect(new CustomClickHouseDialect())
   

3. JDBC方式写入
   确保ClickHouse表字段类型与映射后的Nested类型一致，然后执行写入：

   df.write
       .format("jdbc")
       .option("url", "jdbc:clickhouse://your-clickhouse-host:8123/your_database")
       .option("driver", "com.clickhouse.jdbc.ClickHouseDriver")
       .option("user", "your-username")
       .option("password", "your-password")
       .option("dbtable", "your_table")
       .mode("append")
       .save()
   

内容的提问来源于stack exchange，提问作者Sajal