1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
    <!-- 指定仓库位置，依次为aliyun、cloudera和jboss仓库 -->
    <repositories>
        <repository>
            <id>aliyun</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </repository>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
        <repository>
            <id>jboss</id>
            <url>http://repository.jboss.com/nexus/content/groups/public</url>
        </repository>
    </repositories>

    <!-- 版本属性 -->
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <clickhouse>0.2.4</clickhouse>
    </properties>

    <dependencies>
        <!-- Clickhouse -->
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>${clickhouse}</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
package cn.itcast.clickhouse;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * 编写JDBC代码，从ClickHouse表查询分析数据
	 * step1. 加载驱动类
	 * step2. 获取连接Connection
	 * step3. 创建PreparedStatement对象
	 * step4. 查询数据
	 * step5. 获取数据
	 * step6. 关闭连接
 */
public class ClickHouseJDBCDemo {

	public static void main(String[] args) throws Exception{
		// 定义变量
		Connection conn = null ;
		PreparedStatement pstmt = null ;
		ResultSet result = null ;

		try {
			//step1. 加载驱动类
			Class.forName("ru.yandex.clickhouse.ClickHouseDriver");

			//step2. 获取连接Connection
			conn = DriverManager.getConnection(
				"jdbc:clickhouse://node2.itcast.cn:8123", "root", "123456"
			) ;

			//step3. 创建PreparedStatement对象
			pstmt = conn.prepareStatement("select count(1) from default.ontime");

			//step4. 查询数据
			result = pstmt.executeQuery();

			//step5. 获取数据
			while (result.next()){
				System.out.println(result.getLong(1));
			}
		}catch (Exception e){
			e.printStackTrace();
		}finally {
			//step6. 关闭连接
			if(null != result) result.close();
			if(null != pstmt) pstmt.close();
			if(null != conn) conn.close();
		}
	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
    <repositories>
        <repository>
            <id>mvnrepository</id>
            <url>https://mvnrepository.com/</url>
            <layout>default</layout>
        </repository>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
        <repository>
            <id>elastic.co</id>
            <url>https://artifacts.elastic.co/maven</url>
        </repository>
    </repositories>

    <properties>
        <!--- Scala -->
        <scala.version>2.11.12</scala.version>
        <scala.binary.version>2.11</scala.binary.version>
        <!-- Spark -->
        <spark.version>2.4.0-cdh6.2.1</spark.version>
        <!-- Hadoop -->
        <hadoop.version>3.0.0-cdh6.2.1</hadoop.version>
        <!-- ClickHouse -->
        <clickhouse.version>0.2.4</clickhouse.version>
    </properties>

    <dependencies>
        <!-- 依赖Scala语言 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- Spark Core 依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!-- Spark -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <!-- Hadoop Client 依赖 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <!-- Clickhouse -->
        <dependency>
            <groupId>ru.yandex.clickhouse</groupId>
            <artifactId>clickhouse-jdbc</artifactId>
            <version>${clickhouse.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-databind</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.fasterxml.jackson.core</groupId>
                    <artifactId>jackson-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <outputDirectory>target/classes</outputDirectory>
        <testOutputDirectory>target/test-classes</testOutputDirectory>
        <resources>
            <resource>
                <directory>${project.basedir}/src/main/resources</directory>
            </resource>
        </resources>
        <!-- Maven 编译的插件 -->
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
log4j.rootLogger=${root.logger}
root.logger=WARN,console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n
shell.log.level=WARN
log4j.logger.org.spark-project.jetty=WARN
log4j.logger.org.spark-project.jetty.util.component.AbstractLifeCycle=ERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO
log4j.logger.org.apache.parquet=ERROR
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR
log4j.logger.org.apache.spark.repl.Main=${shell.log.level}
log4j.logger.org.apache.spark.api.python.PythonGatewayServer=${shell.log.level}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
package cn.itcast.clickhouse

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * SparkSQL加载JSON格式文件数据，依据Schema信息在ClickHouse中创建表，并进行数据CUD操作
 */
object ClickHouseSparkDemo {
	
	def main(args: Array[String]): Unit = {
		// 1. 构建SparkSession实例对象
		val spark: SparkSession = SparkSession.builder()
    		.appName(this.getClass.getSimpleName.stripSuffix("$"))
    		.master("local[2]")
    		.config("spark.sql.shuffle.partitions", "2")
    		.getOrCreate()
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// 2. 加载JSON数据：交易订单数据
		val ordersDF: DataFrame = spark.read.json("datas/order.json")
		/*
			root
			 |-- areaName: string (nullable = true)
			 |-- category: string (nullable = true)
			 |-- id: long (nullable = true)
			 |-- money: string (nullable = true)
			 |-- timestamp: string (nullable = true)
		 */
		//ordersDF.printSchema()
		/*
			+--------+--------+---+-----+--------------------+
			|areaName|category|id |money|timestamp           |
			+--------+--------+---+-----+--------------------+
			|北京    |平板电脑|1  |1450 |2019-05-08T01:03.00Z|
			|北京    |手机    |2  |1450 |2019-05-08T01:01.00Z|
			|北京    |手机    |3  |8412 |2019-05-08T01:03.00Z|
			|上海    |电脑    |4  |1513 |2019-05-08T05:01.00Z|
			+--------+--------+---+-----+--------------------+
		 */
		// ordersDF.show(10, truncate = false)
		
		// 3. 依据DataFrame数据集，在ClickHouse数据库中创建表和删除表
		
		
		// 4. 保存DataFrame数据集到ClickHouse表中
		
		
		
		// 5. 更新数据到ClickHouse表中
		val updateDF: DataFrame = Seq(
			(3, 9999, "2020-12-08T01:03.00Z"),
			(4, 9999, "2020-12-08T01:03.00Z")
		).toDF("id", "money", "timestamp")
		
		
		// 6. 删除ClickHouse表中数据
		val deleteDF: DataFrame = Seq( Tuple1(1), Tuple1(2), Tuple1(3)).toDF("id")
		
		// 应用结束，关闭资源
		spark.stop()
	}
	
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package cn.itcast.clickhouse

import org.apache.spark.sql.DataFrame

/**
 * ClickHouse 工具类，创建表、删除表及对表数据进行CUD操作
 */
object ClickHouseUtils extends Serializable {
	

	/**
	 * 创建ClickHouse的连接实例，返回连接对象，通过连接池获取连接对象
	 *
	 * @param host ClickHouse 服务主机地址
	 * @param port ClickHouse 服务端口号
	 * @param username 用户名称，默认用户：default
	 * @param password 密码，默认为空
	 * @return Connection 对象
	 */
	def createConnection(host: String, port: String,
	                     username: String, password: String): ClickHouseConnection = {
        // 加载驱动类
		Class.forName("ru.yandex.clickhouse.ClickHouseDriver")
		// TODO: 使用ClickHouse中提供ClickHouseDataSource获取连接对象
		val datasource: ClickHouseDataSource = new ClickHouseDataSource(s"jdbc:clickhouse://${host}:${port}")
		// 获取连接对象
		val connection: ClickHouseConnection = datasource.getConnection(username, password)
		// 返回对象
		connection
	}
    
}    

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
	/**
	 * 插入数据：DataFrame到ClickHouse表
	 */
	def insertData(dataframe: DataFrame,
	               dbName: String, tableName: String): Unit = {
		
	}
	
	/**
	 * 更新数据：依据主键，更新DataFrame数据到ClickHouse表
	 */
	def updateData(dataframe: DataFrame, dbName: String,
	               tableName: String, primaryField: String = "id"): Unit = {
		
	}
	
	/**
	 * 删除数据：依据主键，将ClickHouse表中数据删除
	 */
	def deleteData(dataframe: DataFrame, dbName: String,
	               tableName: String, primaryField: String = "id"): Unit = {
		
	}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
	/**
	 * 传递Connection对象和SQL语句对ClickHouse发送请求，执行更新操作
	 *
	 * @param sql 执行SQL语句或者DDL语句
	 */
	def executeUpdate(sql: String): Unit = {
		// 声明变量
		var conn: ClickHouseConnection = null
		var pstmt: ClickHouseStatement = null
		try{
			// a. 获取ClickHouse连接对象
			conn = ClickHouseUtils.createConnection("node2.itcast.cn", "8123", "root", "123456")
			// b. 获取PreparedStatement实例对象
			pstmt = conn.createStatement()
			// c. 执行更新操作
			pstmt.executeUpdate(sql)
		}catch {
			case e: Exception => e.printStackTrace()
		}finally {
			// 关闭连接
			if (null != pstmt) pstmt.close()
			if (null != conn) conn.close()
		}
	}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
package cn.itcast.clickhouse.spark

import org.apache.spark.sql.DataFrame
import ru.yandex.clickhouse.{ClickHouseConnection, ClickHouseDataSource, ClickHouseStatement}

/**
 * 工具类，对ClickHouse数据库进行DDL（创建表和删除表）和DML（CUD）操作方法封装。
 */
object ClickHouseUtils extends Serializable {
	
	/**
	 * 创建ClickHouse的连接实例，返回连接对象，通过连接池获取连接对象
	 *
	 * @param host ClickHouse 服务主机地址
	 * @param port ClickHouse 服务端口号
	 * @param username 用户名称，默认用户：default
	 * @param password 密码，默认为空
	 * @return Connection 对象
	 */
	def createConnection(host: String, port: String,
	                     username: String, password: String): ClickHouseConnection = {
		// 加载驱动类
		Class.forName("ru.yandex.clickhouse.ClickHouseDriver")
		// TODO: 使用ClickHouse中提供ClickHouseDataSource获取连接对象
		val datasource: ClickHouseDataSource = new ClickHouseDataSource(s"jdbc:clickhouse://${host}:${port}")
		// 获取连接对象
		val connection: ClickHouseConnection = datasource.getConnection(username, password)
		// 返回对象
		connection
	}
	
	/* ================  TODO： DML操作，封装方法 =============================== */
	/**
	 * 插入数据：DataFrame到ClickHouse表
	 */
	def insertData(dataframe: DataFrame, dbName: String, tableName: String): Unit = {
	
	}
	
	/**
	 * 更新数据：依据主键，更新DataFrame数据到ClickHouse表
	 */
	def updateData(dataframe: DataFrame, dbName: String, tableName: String, primaryField: String = "id"): Unit = {
	
	}
	
	/**
	 * 删除数据：依据主键，将ClickHouse表中数据删除
	 */
	def deleteData(dataframe: DataFrame, dbName: String, tableName: String, primaryField: String = "id"): Unit = {
	
	}
	
	/* ================  TODO： DDL操作，封装方法 =============================== */
	// 在数据库ClickHouse或者MySQL中，创建表和删除表，执行DDL语句即可（CREATE 语句和DROP 语句）
	/**
	 * 传递Connection对象和SQL语句对ClickHouse发送请求，执行更新操作
	 *
	 * @param sql 执行SQL语句或者DDL语句
	 */
	def executeUpdate(sql: String): Unit = {
		// 声明变量
		var conn: ClickHouseConnection = null
		var pstmt: ClickHouseStatement = null
		try{
			// a. 获取ClickHouse连接对象
			conn = ClickHouseUtils.createConnection("node2.itcast.cn", "8123", "root", "123456")
			// b. 获取PreparedStatement实例对象
			pstmt = conn.createStatement()
			// c. 执行更新操作：可以是DDL语句
			pstmt.executeUpdate(sql)
		}catch {
			case e: Exception => e.printStackTrace()
		}finally {
			// 关闭连接
			if (null != pstmt) pstmt.close()
			if (null != conn) conn.close()
		}
	}
	
	
}

1
2
3
4
5
6
7
8
9
10
11
12
CREATE TABLE IF NOT EXISTS test.tbl_order (
areaName String,
category String,
id Int64,
money String,
timestamp String,
sign Int8,
version Int8
)
ENGINE=VersionedCollapsingMergeTree(sign, version)
ORDER BY id ;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
	/**
	 * 依据DataFrame数据集中约束Schema信息，构建ClickHouse表创建DDL语句
	 *
	 * @param dbName 数据库的名称
	 * @param tableName 表的名称
	 * @param schema DataFrame数据集约束Schema
	 * @param primaryKeyField ClickHouse表主键字段名称
	 * @return ClickHouse表创建DDL语句
	 */
	def createTableDdl(dbName: String, tableName: String,
	                   schema: StructType, primaryKeyField: String = "id"): String = {
		/*
		areaName String,
		category String,
		id Int64,
		money String,
		timestamp String,
		 */
		val fieldStr: String = schema.fields.map{field =>
			// 获取字段名称
			val fieldName: String = field.name
			// 获取字段类型
			val fieldType: String = field.dataType match {
				case StringType => "String"
				case IntegerType => "Int32"
				case FloatType => "Float32"
				case LongType => "Int64"
				case BooleanType => "UInt8"
				case DoubleType => "Float64"
				case DateType => "Date"
				case TimestampType => "DateTime"
				case x => throw new Exception(s"Unsupported type: ${x.toString}")
			}
			// 组合字符串
			s"${fieldName} ${fieldType}"
		}.mkString(",n")
	
		// 构建创建表DDL语句
		val createDdl: String = s"""
		  |CREATE TABLE IF NOT EXISTS ${dbName}.${tableName} (
		  |${fieldStr},
		  |sign Int8,
		  |version UInt64
		  |)
		  |ENGINE=VersionedCollapsingMergeTree(sign, version)
		  |ORDER BY ${primaryKeyField}
		  |""".stripMargin
		
		// 返回DDL语句
		createDdl
	}

1
2
DROP TABLE IF EXISTS test.tbl_orders ;

1
2
3
4
5
6
7
8
9
10
11
	/**
	 * 依据数据库名称和表名称，构建ClickHouse中删除表DDL语句
	 *
	 * @param dbName 数据库名称
	 * @param tableName 表名称
	 * @return 删除表的DDL语句
	 */
	def dropTableDdl(dbName: String, tableName: String): String = {
		s"DROP TABLE IF EXISTS ${dbName}.${tableName}"
	}

1
2
INSERT INTO test.tbl_order (areaName, category, id, money, timestamp, sign, version) VALUES ('北京', '平板电脑', 1, '1450', '2019-05-08T01:03.00', 1, 1);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
	/**
	 * 构建数据插入ClickHouse中SQL语句
	 *
	 * @param dbName 数据库名称
	 * @param tableName 表的名称
	 * @param columns 列名称
	 * @return INSERT 插入语句
	 */
	def createInsertSQL(dbName: String, tableName: String, columns: Array[String] ): String = {
		// 所有列名称字符串 -> areaName, category, id, money, timestamp
		val fieldsStr: String = columns.mkString(", ")
		// 所有列对应值的占位符 -> ?, ?, ?, ?, ?
		val valuesStr: String = columns.map(_ => "?").mkString(", ")
		// 插入INSERT SQL语句
		s"""
		   |INSERT INTO ${dbName}.${tableName} (${fieldsStr}, sign, version) VALUES (${valuesStr}, 1, 1);
		   |""".stripMargin
	}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
	/**
	 * 插入数据：DataFrame到ClickHouse表
	 */
	def insertData(dataframe: DataFrame, dbName: String, tableName: String): Unit = {
		// 获取DataFrame中Schema信息
		val schema: StructType = dataframe.schema
		// 获取DataFrame中所有列名称
		val columns: Array[String] = dataframe.columns
		
		// 构建INSERT语句
		val insertSql: String = createInsertSQL(dbName, tableName, columns)
		// TODO：针对每个分区数据进行操作，每个分区的数据进行批量插入
		dataframe.foreachPartition{iter =>
			// 声明变量
			var conn: ClickHouseConnection = null
			var pstmt: PreparedStatement = null
			try{
				// a. 获取ClickHouse连接对象
				conn = ClickHouseUtils.createConnection("node2.itcast.cn", "8123", "root", "123456")
				// b. 构建PreparedStatement对象
				pstmt = conn.prepareStatement(insertSql)
				// TODO: c. 遍历每个分区中数据，将每条数据Row值进行设置
				var counter: Int = 0
				iter.foreach{row =>
					// 从row中获取每列的值，和索引下标 -> 通过列名称 获取 Row中下标索引 , 再获取具体值
					columns.foreach{column =>
						// 通过列名称 获取 Row中下标索引
						val index: Int = schema.fieldIndex(column)
						// 依据索引下标，从Row中获取值
						val value: Any = row.get(index)
						// 进行PreparedStatement设置值
						pstmt.setObject(index + 1, value)
					}
					pstmt.setObject(columns.length + 1, 1)
					pstmt.setObject(columns.length + 2, 1)
					
					// 加入批次
					pstmt.addBatch()
					counter += 1
					// 判断counter大小，如果大于1000 ，进行一次批量插入
					if(counter >= 1000) {
						pstmt.executeBatch()
						counter = 0
					}
				}
				// d. 批量插入
				pstmt.executeBatch()
			}catch {
				case e: Exception => e.printStackTrace()
			}finally {
				// e. 关闭连接
				if(null != pstmt) pstmt.close()
				if(null != conn) conn.close()
			}
		}
	}

1
2
ALTER TABLE test.tbl_order UPDATE money = '9999', timestamp = '2020-12-08T01:03.00Z' WHERE id = 3 ;

1
2
3
4
5
val updateDF: DataFrame = Seq(
			(3, 9999, "2020-12-08T01:03.00Z"),
			(4, 9999, "2020-12-08T01:03.00Z")
		).toDF("id", "money", "timestamp")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
	/**
	 * 将DataFrame数据集更新至ClickHouse表中，依据主键进行更新
	 *
	 * @param dbName 数据库名称
	 * @param tableName 表名称
	 * @param dataframe 数据集DataFrame，更新的数据，其中要包含主键
	 */
	def updateData(dataframe: DataFrame, dbName: String,
	               tableName: String, primaryField: String = "id"): Unit = {
		// 对DataFrame每个分区数据进行操作
		dataframe.foreachPartition{iter =>
			// 声明变量
			var conn: ClickHouseConnection = null
			var pstmt: ClickHouseStatement = null
			try{
				// a. 获取ClickHouse连接对象
				conn = ClickHouseUtils.createConnection("node2.itcast.cn", "8123", "root", "123456")
				// b. 构建PreparedStatement对象
				pstmt = conn.createStatement()
				
				// TODO: 遍历分区中每条数据Row，构建更新Update语句，进行更新操作
				iter.foreach{row =>
					// b. 依据Row对象，创建更新SQL语句
					// ALTER TABLE db_ck.tbl_orders UPDATE money = '9999', timestamp = '2020-12-08T01:03.00Z' WHERE id = 3 ;
					val updateSql: String = createUpdateSQL(dbName, tableName, row)
					// d. 执行更新操作
					pstmt.executeUpdate(updateSql)
				}
			}catch {
				case e: Exception => e.printStackTrace()
			}finally {
				// e. 关闭连接
				if(null != pstmt) pstmt.close()
				if(null != conn) conn.close()
			}
		}
	}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
	/**
	 * 构建数据更新ClickHouse表中SQL语句
	 *
	 * @param dbName 数据库名称
	 * @param tableName 表名称
	 * @param row DataFrame中每行数据
	 * @param primaryKeyField 主键列
	 * @return update更新SQL语句
	 */
	def createUpdateSQL(dbName: String, tableName: String,
	                    row: Row, primaryKeyField: String = "id"): String = {
		/*
				 id     money    timestamp
			Row: 3,    9999, "2020-12-08T01:03.00Z"
						     |
			     money = '9999', timestamp = '2020-12-08T01:03.00Z'
		 */
		val updatesStr: String = row.schema.fields
    		.map(field => field.name) // 获取所有列名称
			// 过滤主键列名称
    		.filter(columnName => ! primaryKeyField.equals(columnName))
			// 依据列名称获取对应的值
    		.map{columnName =>
			    val columnValue: Any = getFieldValue(row, columnName)
			    s"${columnName} = '${columnValue}'"
		    }.mkString(", ")
		// 获取主键的值
		val primaryKeyValue: Any = getFieldValue(row, primaryKeyField)
		// 构建UPDATE更新SQL语句
		s"""
		   |ALTER TABLE ${dbName}.${tableName}
		   |    UPDATE ${updatesStr} WHERE ${primaryKeyField} = ${primaryKeyValue} ;
		   |""".stripMargin
	}

1
2
ALTER TABLE test.tbl_order DELETE WHERE id = "3" ;

1
2
val deleteDF: DataFrame = Seq( Tuple1(1), Tuple1(2), Tuple1(3)).toDF("id")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
	/**
	 * 删除数据：依据主键，将ClickHouse表中数据删除
	 */
	def deleteData(dataframe: DataFrame, dbName: String,
	               tableName: String, primaryField: String = "id"): Unit = {
		// 对DataFrame每个分区数据进行操作
		dataframe.foreachPartition{iter =>
			// 声明变量
			var conn: ClickHouseConnection = null
			var pstmt: ClickHouseStatement = null
			try{
				// a. 获取ClickHouse连接对象
				conn = ClickHouseUtils.createConnection("node2.itcast.cn", "8123", "root", "123456")
				// b. 构建PreparedStatement对象
				pstmt = conn.createStatement()
				
				// TODO: 遍历分区中每条数据Row，构建更新Update语句，进行更新操作
				iter.foreach{row =>
					// b. 依据Row对象，创建删除SQL语句
					// ALTER TABLE db_ck.tbl_orders DELETE WHERE id = "3" ;
					val deleteSql: String = s"ALTER TABLE ${dbName}.${tableName} DELETE WHERE ${primaryField} = ${getFieldValue(row, primaryField)} ;"
					// d. 执行更新操作
					pstmt.executeUpdate(deleteSql)
				}
			}catch {
				case e: Exception => e.printStackTrace()
			}finally {
				// e. 关闭连接
				if(null != pstmt) pstmt.close()
				if(null != conn) conn.close()
			}
		}
	}