Kafka与ClickHouse开发与应用_tyt2023

在上步ClickHouseDemo处右键点击，选择“New->Class”创建Java类Util并输入信息如下：

package ClickHouseDemo;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Date;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

public class Util {

private static final String JDBC_PREFIX = "jdbc:clickhouse://";

ArrayList<ArrayList<ArrayList<String>>> exeSql(ArrayList<String> sqlList) throws Exception {
ArrayList<ArrayList<ArrayList<String>>> multiSqlResults = new ArrayList<ArrayList<ArrayList<String>>>();
for (String sql : sqlList) {
ArrayList<ArrayList<String>> singleSqlResult = exeSql(sql);
multiSqlResults.add(singleSqlResult);
}
return multiSqlResults;
}

//数据查询工具函数
ArrayList<ArrayList<String>> exeSql(String sql) throws Exception {
ArrayList<ArrayList<String>> resultArrayList = new ArrayList<ArrayList<String>>();
List<String> serverList = Demo.ckLbServerList;
Connection connection = null;
Statement statement = null;
ResultSet resultSet;
String user = Demo.user;
String password = Demo.password;
try {
Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
ClickHouseProperties clickHouseProperties = new ClickHouseProperties();
clickHouseProperties.setSocketTimeout(60000);
if (Demo.sslUsed) {
clickHouseProperties.setSsl(true);
clickHouseProperties.setSslMode("none");
}
for (int tries = 1; tries <= serverList.size(); tries++) {
try {
System.out.println("Current load balancer is " + serverList.get(tries - 1));
ClickHouseDataSource clickHouseDataSource =
new ClickHouseDataSource(JDBC_PREFIX + serverList.get(tries - 1), clickHouseProperties);
connection = clickHouseDataSource.getConnection(user, password);
statement = connection.createStatement();
System.out.println("Execute query: " + sql);
long begin = System.currentTimeMillis();
resultSet = statement.executeQuery(sql);
long end = System.currentTimeMillis();
if (null != resultSet && null != resultSet.getMetaData()) {
int columnCount = resultSet.getMetaData().getColumnCount();
ArrayList<String> rowResultArray = new ArrayList<String>();
for (int j = 1; j <= columnCount; j++) {
rowResultArray.add(resultSet.getMetaData().getColumnName(j));
}
resultArrayList.add(rowResultArray);
while (resultSet.next()) {
rowResultArray = new ArrayList<String>();
for (int j = 1; j <= columnCount; j++) {
rowResultArray.add(resultSet.getString(j));
}
if (rowResultArray.size() != 0) {
resultArrayList.add(rowResultArray);
}
}
}
} catch (Exception e) {
System.out.println(e.toString());
System.out.println("this is the " + tries);
if (tries == serverList.size()) {
throw e;
}
continue;
}
break;
}
} catch (Exception e) {
System.out.println(e.toString());
throw e;
} finally {
try {
if (statement != null) {
statement.close();
}
if (connection != null) {
connection.close();
}
} catch (SQLException e) {
System.out.println(e.getMessage());
}
}
return resultArrayList;
}

//数据插入工具函数
void insertData(String databaseName, String tableName, int batchNum, int batchRows) throws Exception {
List<String> serverList = Demo.ckLbServerList;
Connection connection = null;
String user = Demo.user;
String password = Demo.password;
try {
Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
ClickHouseProperties clickHouseProperties = new ClickHouseProperties();
clickHouseProperties.setSocketTimeout(60000);
if (Demo.sslUsed) {
clickHouseProperties.setSsl(true);
clickHouseProperties.setSslMode("none");
}
for (int tries = 1; tries <= serverList.size(); tries++) {
try {
System.out.println("Current load balancer is " + serverList.get(tries - 1));
ClickHouseDataSource clickHouseDataSource =
new ClickHouseDataSource(JDBC_PREFIX + serverList.get(tries - 1), clickHouseProperties);
connection = clickHouseDataSource.getConnection(user, password);
String insertSql = "insert into " + databaseName + "." + tableName + " values (?,?,?)";
PreparedStatement preparedStatement = connection.prepareStatement(insertSql);
long allBatchBegin = System.currentTimeMillis();
for (int j = 0; j < batchNum; j++) {
for (int i = 0; i < batchRows; i++) {
preparedStatement.setString(1, "huawei_" + (i + j * 10));
preparedStatement.setInt(2, ((int) (Math.random() * 100)));
preparedStatement.setDate(3, generateRandomDate("2018-01-01", "2021-12-31"));
preparedStatement.addBatch();
}
long begin = System.currentTimeMillis();
preparedStatement.executeBatch();
long end = System.currentTimeMillis();
}
long allBatchEnd = System.currentTimeMillis();
} catch (Exception e) {
System.out.println(e.toString());
System.out.println("this is the " + tries);
if (tries == serverList.size()) {
throw e;
}
continue;
}
break;
}
} catch (Exception e) {
System.out.println(e.getMessage());
throw e;
} finally {
try {
if (connection != null) {
connection.close();
}
} catch (SQLException e) {
System.out.println(e.getMessage());
}
}
}

//测试数据生成工具函数
private Date generateRandomDate(String beginDate, String endDate) {
try {
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
java.util.Date start = format.parse(beginDate);// \u6784\u9020\u5f00\u59cb\u65e5\u671f
java.util.Date end = format.parse(endDate);// \u6784\u9020\u7ed3\u675f\u65e5\u671f
// getTime()\u8868\u793a\u8fd4\u56de\u81ea 1970 \u5e74 1 \u6708 1 \u65e5 00:00:00 GMT \u4ee5\u6765\u6b64 Date \u5bf9\u8c61\u8868\u793a\u7684\u6beb\u79d2\u6570\u3002
if (start.getTime() >= end.getTime()) {
return null;
}
long date = random(start.getTime(), end.getTime());
return new Date(date);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}

//测试数据生成工具函数
private static long random(long begin, long end) {
long rtn = begin + (long) (Math.random() * (end - begin));
// \u5982\u679c\u8fd4\u56de\u7684\u662f\u5f00\u59cb\u65f6\u95f4\u548c\u7ed3\u675f\u65f6\u95f4\uff0c\u5219\u9012\u5f52\u8c03\u7528\u672c\u51fd\u6570\u67e5\u627e\u968f\u673a\u503c
if (rtn == begin || rtn == end) {
return random(begin, end);
}
return rtn;
}
}

 程序打包

scp /home/user/eclipse-workspace/ClickHouseExample/target/ClickHouseExample-0.0.1-SNAPSHOT-jar-with-dependencies.jar root@xxx.xxx.xxx.xxx:/home/omm

scp -r /home/user/eclipse-workspace/ClickHouseExample/conf root@xxx.xxx.xxx.xxx:/home/omm