海量数据存储ClickHouse

docker run -d --name ybchen_clickhouse --ulimit nofile=262144:262144 \
-p 8123:8123 -p 9000:9000 -p 9009:9009 --privileged=true \
-v /mydata/docker/clickhouse/log:/var/log/clickhouse-server \
-v /mydata/docker/clickhouse/data:/var/lib/clickhouse clickhouse/clickhouse-server:22.2.3.5

CREATE DATABASE shop

CREATE TABLE shop.clickstream (
customer_id String,
time_stamp Date,
click_event_type String,
page_code FixedString(20),
source_id UInt64
)
ENGINE = MergeTree()
ORDER BY (time_stamp)

INSERT INTO shop.clickstream
VALUES ('customer1', '2021-10-02', 'add_to_cart', 'home_enter', 568239 )

SELECT * FROM shop.clickstream

SELECT * FROM shop.clickstream WHERE time_stamp >= '2001-11-01'

有符号整型范围
Int8 — [-128 : 127]Int16 — [-32768 : 32767]Int32 — [-2147483648 : 2147483647]Int64 — [-9223372036854775808 : 9223372036854775807]Int128 — [-170141183460469231731687303715884105728 : 170141183460469231731687303715884105727]Int256 — [-57896044618658097711785492504343953926634992332820282019728792003956564819968 : 57896044618658097711785492504343953926634992332820282019728792003956564819967]无符号整型范围
UInt8 — [0 : 255]UInt16 — [0 : 65535]UInt32 — [0 : 4294967295]UInt64 — [0 : 18446744073709551615]UInt128 — [0 : 340282366920938463463374607431768211455]UInt256 — [0 : 115792089237316195423570985008687907853269984665640564039457584007913129639935]

CREATE TABLE shop.clickstream1 (
customer_id String,
time_stamp Date,
click_event_type String,
page_code FixedString(20),
source_id UInt64,
money Decimal(2,1)
)
ENGINE = MergeTree()
ORDER BY (time_stamp)INSERT INTO shop.clickstream1
VALUES ('customer1', '2021-10-02', 'add_to_cart', 'home_enter', 568239,2.11 )

61f0c404-5cb3-11e7-907b-a6006ad3dba0

00000000-0000-0000-0000-000000000000

CREATE TABLE t_uuid (x UUID, y String) ENGINE=TinyLogINSERT INTO t_uuid SELECT generateUUIDv4(), 'Example 1'

CREATE TABLE t_enum
(
page_code Enum8('home' = 1, 'detail' = 2,'pay'=3)
)
ENGINE = TinyLog

#插入成功
INSERT INTO t_enum VALUES ('home'), ('detail')#插入报错
INSERT INTO t_enum VALUES ('home1')#查询
SELECT * FROM t_enum

CREATE TABLE shop.clickstream2 (
customer_id String,
time_stamp Date,
click_event_type String,
page_code FixedString(20),
source_id UInt64,
money Decimal(2,1),
is_new Bool
)
ENGINE = MergeTree()
ORDER BY (time_stamp)DESCRIBE shop.clickstream2INSERT INTO shop.clickstream2
VALUES ('customer1', '2021-10-02', 'add_to_cart', 'home_enter', 568239, 3.8,1)

select * from system.data_type_families

CREATE TABLE shop.clickstream3 (
customer_id String,
time_stamp Date,
click_event_type String,
page_code FixedString(20),
source_id UInt64,
money Decimal(2,1),
is_new Bool
)
ENGINE = MergeTree()
ORDER BY (time_stamp)

DESCRIBE shop.clickstream3

SELECT * FROM shop.clickstream3

INSERT INTO shop.clickstream3
VALUES ('customer2', '2021-10-02', 'add_to_cart', 'home_enter', 568239,2.1, False )

SELECT database, table, command, create_time, is_done FROM system.mutations LIMIT 20

ALTER TABLE shop.clickstream3 UPDATE click_event_type = 'pay' where customer_id = 'customer2';

ALTER TABLE shop.clickstream3 delete where customer_id = 'customer2';

create table shop.order_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =MergeTree()
partition by toYYYYMMDD(create_time)
order by (id,sku_id)
primary key (id);

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1] [TTL expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2] [TTL expr2],
...
INDEX index_name1 expr1 TYPE type1(...) GRANULARITY value1,
INDEX index_name2 expr2 TYPE type2(...) GRANULARITY value2
) ENGINE = MergeTree()
ORDER BY expr
[PARTITION BY expr]
[PRIMARY KEY expr]
[SAMPLE BY expr]
[TTL expr [DELETE|TO DISK 'xxx'|TO VOLUME 'xxx'], ...]
[SETTINGS name=value, ...]

create table shop.order_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =MergeTree()
order by (id,sku_id)
partition by toYYYYMMDD(create_time)
primary key (id);

insert into shop.order_merge_tree values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_1','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_2','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_4','65432',600.01,'2023-04-02 11:00:00');

docker exec -it 35dad2d981d5 /bin/bash

新的数据写入会有临时分区产生，不之间加入已有分区写入完成后经过一定时间（10到15分钟），ClickHouse会自动化执行合并操作，将临时分区的数据合并到已有分区当中optimize的合并操作是在后台执行的，无法预测具体执行时间点，除非是手动执行通过手工合并( optimize table xxx final; ) 在数据量比较大的情况，尽量不要使用该命令，执行optimize要消耗大量时间

create table shop.order_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =MergeTree()
order by (id,sku_id)
partition by toYYYYMMDD(create_time)
primary key (id);insert into shop.order_merge_tree values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_1','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_2','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_4','65432',600.01,'2023-04-02 11:00:00');docker exec -it 35dad2d981d5 /bin/bash clickhouse-client optimize table shop.order_merge_tree final;

create table shop.order_merge_tree22(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =MergeTree()
order by (id,sku_id) primary key (id); insert into shop.order_merge_tree22 values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_1','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_2','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_4','65432',600.01,'2023-04-02 11:00:00');

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = ReplacingMergeTree([ver])
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]

create table shop.order_relace_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =ReplacingMergeTree(id)
order by (sku_id)
partition by toYYYYMMDD(create_time)
primary key (sku_id);

insert into shop.order_relace_merge_tree values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_5','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_6','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_7','65432',600.01,'2023-04-02 11:00:00');insert into shop.order_relace_merge_tree values
(11,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(21,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(31,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(41,'sku_5','54222',2000.3,'2023-04-01 19:00:00'),
(51,'sku_8','53423',120.2,'2023-04-01 19:00:00'),
(61,'sku_9','65432',600.01,'2023-04-02 11:00:00');

docker exec -it 35dad2d981d5 /bin/bash clickhouse-clientSELECT * FROM shop.order_relace_merge_tree optimize table shop.order_relace_merge_tree final;

create table shop.order_relace_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =ReplacingMergeTree(id)
order by (sku_id)
partition by toYYYYMMDD(create_time)
primary key (sku_id);insert into shop.order_relace_merge_tree values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_5','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_6','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_7','65432',600.01,'2023-04-02 11:00:00');insert into shop.order_relace_merge_tree values
(11,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(21,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(31,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(41,'sku_5','54222',2000.3,'2023-04-01 19:00:00'),
(51,'sku_8','53423',120.2,'2023-04-01 19:00:00'),
(61,'sku_9','65432',600.01,'2023-04-02 11:00:00');docker exec -it 35dad2d981d5 /bin/bash clickhouse-clientSELECT * FROM shop.order_relace_merge_tree optimize table shop.order_relace_merge_tree final;

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = SummingMergeTree([columns])
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]

create table shop.order_summing_merge_tree(
id UInt32,
sku_id String,
out_trade_no String,
total_amount Decimal(16,2),
create_time Datetime
) engine =SummingMergeTree(total_amount)
order by (id,sku_id)
partition by toYYYYMMDD(create_time)
primary key (id);

insert into shop.order_summing_merge_tree values
(1,'sku_1','aabbcc',5600.00,'2023-03-01 16:00:00') ,
(2,'sku_2','23241',4.02,'2023-03-01 17:00:00'),
(3,'sku_3','542323',55.02,'2023-03-01 18:00:00'),
(4,'sku_5','54222',2000.3,'2023-04-01 19:00:00'),
(5,'sku_6','53423',120.2,'2023-04-01 19:00:00'),
(6,'sku_7','65432',600.01,'2023-04-02 11:00:00');insert into shop.order_summing_merge_tree values
(1,'sku_1','aabbccbb',5600.00,'2023-03-01 23:09:00')

select sku_id,sum(total_amount) from shop.order_summing_merge_tree group by sku_id

optimize table shop.order_summing_merge_tree final;

docker run -d --name ybchen_zookeeper -p 2181:2181 -t zookeeper:3.7.0

#各个节点上传到新建文件夹
/usr/local/software/*#安装
sudo rpm -ivh *.rpm#启动
systemctl start clickhouse-server#停止
systemctl stop clickhouse-server#重启
systemctl restart clickhouse-server#状态查看
sudo systemctl status clickhouse-server#查看端口占用，如果命令不存在 yum install -y lsof
lsof -i :8123#查看日志
tail -f /var/log/clickhouse-server/clickhouse-server.log
tail -f /var/log/clickhouse-server/clickhouse-server.err.log#开启远程访问，取消下面的注释
vim /etc/clickhouse-server/config.xml#编辑配置文件
<listen_host>0.0.0.0</listen_host>#重启
systemctl restart clickhouse-server# 增加dns解析
sudo vim /etc/hosts172.16.0.103 ybchen-1
172.16.0.108 ybchen-2

#进入配置目录
cd /etc/clickhouse-server#编辑配置文件
sudo vim config.xml#找到zookeeper节点，增加下面的，如果有多个zk则按照顺序加即可<zookeeper>
<node>
<host>172.16.0.103</host>
<port>2181</port>
</node>
</zookeeper>#重启
systemctl restart clickhouse-server########建表#######
#节点一，zk路径一致，副本名称不一样
CREATE TABLE tb_product
(
userId UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/tb_product', 'product-replica-1')
ORDER BY (userId)#节点二，zk路径一致，副本名称不一样
CREATE TABLE tb_product
(
userId UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/1/tb_product', 'product-replica-2')
ORDER BY (userId)注意
副本只能同步数据，不能同步表结构，需要在每台机器上手动建表
#插入和查询数据
INSERT into tb_product values(1),(2),(3)select * from tb_product#查询zk配置
select * from system.zookeeper where path='/'

CREATE TABLE tb_order
(
EventDate DateTime,
CounterID UInt32,
UserID UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/01/tb_order', 'tb_order_01')
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID))
SAMPLE BY intHash32(UserID)

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]])
[SETTINGS name=value, ...]

<remote_servers>
<logs>
<!-- 分布式查询的服务器间集群密码
默认值:无密码(将不执行身份验证) 如果设置了，那么分布式查询将在分片上验证，所以至少:
- 这样的集群应该存在于shard上
- 这样的集群应该有相同的密码。 而且(这是更重要的)，initial_user将作为查询的当前用户使用。
-->
<!-- <secret></secret> -->
<shard>
<!-- 可选的。写数据时分片权重。 默认: 1. -->
<weight>1</weight>
<!-- 可选的。是否只将数据写入其中一个副本。默认值:false(将数据写入所有副本)。 -->
<internal_replication>false</internal_replication>
<replica>
<!-- 可选的。负载均衡副本的优先级，请参见（load_balancing 设置)。默认值:1(值越小优先级越高)。 -->
<priority>1</priority>
<host>example01-01-1</host>
<port>9000</port>
</replica>
<replica>
<host>example01-01-2</host>
<port>9000</port>
</replica>
</shard>
<shard>
<weight>2</weight>
<internal_replication>false</internal_replication>
<replica>
<host>example01-02-1</host>
<port>9000</port>
</replica>
<replica>
<host>example01-02-2</host>
<secure>1</secure>
<port>9440</port>
</replica>
</shard>
</logs>
</remote_servers>

#进入配置目录
cd /etc/clickhouse-server#编辑配置文件
sudo vim config.xml<!-- 2shard 1replica -->
<cluster_2shards_1replicas>
<!-- shard1 -->
<shard>
<replica>
<host>172.16.0.103</host>
<port>9000</port>
</replica>
</shard> <!-- shard2 -->
<shard>
<replica>
<host>172.16.0.108</host>
<port> 9000</port>
</replica>
</shard> </cluster_2shards_1replicas> <zookeeper>
<node >
<host>172.16.0.103</host>
<port>2181</port>
</node>
</zookeeper>

systemctl restart clickhouse-server

select * from system.clusters

#【选一个节点】创建好本地表后，在1个节点创建，会在其他节点都存在
create table default.ybchen_order on cluster cluster_2shards_1replicas
(id Int8,name String) engine =MergeTree order by id;#【选一个节点】创建分布式表名 ybchen_order_all，在1个节点创建，会在其他节点都存在
create table ybchen_order_all on cluster cluster_2shards_1replicas (
id Int8,name String
)engine = Distributed(cluster_2shards_1replicas,default, ybchen_order,hiveHash(id));#分布式表插入
insert into ybchen_order_all values(1,'老陈'),(2,'老王'),(3,'老李'),(4,'老赵');#【任意节点查询-分布式，全部数据】
SELECT * from ybchen_order_all#【任意本地节点查询，部分数据】
SELECT * from ybchen_order

CREATE TABLE default.visit_stats
(
`product_id` UInt64,
`is_new` UInt16,
`province` String,
`city` String,
`pv` UInt32,
`visit_time` DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(visit_time)
ORDER BY (
product_id,
is_new,
province,
city
);

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.13</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.ybchen</groupId>
<artifactId>ybchen-clickhouse</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>ybchen-clickhouse</name>
<description>SpringBoot整合Clickhouse</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency> <!-- lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency> <!-- clickhouse -->
<dependency>
<groupId>ru.yandex.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.3.2</version>
</dependency> <!--mybatis plus-->
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
<version>3.5.1</version>
</dependency>
</dependencies> <build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build></project>

server.port=8888
spring.datasource.driver-class-name=ru.yandex.clickhouse.ClickHouseDriver
spring.datasource.url=jdbc:clickhouse://47.116.143.16:8123/default
# 账号
#spring.datasource.username=
# 密码
#spring.datasource.password=mybatis-plus.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl
logging.level.root=INFO

package com.ybchen.model;import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;/**
* @Description:
* @Author：chenyanbin
* @CreateTime: 2022-05-14 21:51
* @Version：1.0.0
*/
@Data
@TableName("visit_stats")
public class VisitStatsDO {
/**
* 商品
*/
private Long productId; /**
* 1是新访客，0是老访客
*/
private Integer isNew; /**
* 省份
*/
private String province; /**
* 城市
*/
private String city; /**
* 访问量
*/
private Integer pv; /**
* 访问时间
*/
private String visitTime;
}

package com.ybchen.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.ybchen.model.VisitStatsDO;public interface VisitStatsMapper extends BaseMapper<VisitStatsDO> {
}

package com.ybchen.controller;import com.ybchen.request.VisitRecordListRequest;
import com.ybchen.service.VisitService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.Map;/**
* @Description:
* @Author：chenyanbin
* @CreateTime: 2022-05-14 21:53
* @Version：1.0.0
*/
@RestController
@RequestMapping("/api/v1/data")
public class DataController {
@Autowired
VisitService visitService; @PostMapping("list")
public Map<String,Object> queryVisitRecordList(
@RequestBody VisitRecordListRequest request
) {
return visitService.queryVisitRecordList(request);
}
}

package com.ybchen.request;import lombok.Data;/**
* @Description:
* @Author：chenyanbin
* @CreateTime: 2022-05-14 21:54
* @Version：1.0.0
*/
@Data
public class VisitRecordListRequest {
private Long productId;
private int page;
private int size;
}

package com.ybchen.service;import com.ybchen.request.VisitRecordListRequest;import java.util.Map;public interface VisitService {
/**
* 查询访问记录
* @param request
* @return
*/
Map<String, Object> queryVisitRecordList(VisitRecordListRequest request);
}

package com.ybchen.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ybchen.mapper.VisitStatsMapper;
import com.ybchen.model.VisitStatsDO;
import com.ybchen.request.VisitRecordListRequest;
import com.ybchen.service.VisitService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.util.HashMap;
import java.util.Map;/**
* @Description:
* @Author：chenyanbin
* @CreateTime: 2022-05-14 21:55
* @Version：1.0.0
*/
@Service
@Slf4j
public class VisitServiceImpl implements VisitService {
@Autowired
VisitStatsMapper visitStatsMapper; @Override
public Map<String, Object> queryVisitRecordList(VisitRecordListRequest request) {
Map<String, Object> dataMap = new HashMap<>(3);
IPage<VisitStatsDO> pageInfo = new Page<>(request.getPage(), request.getSize());
IPage<VisitStatsDO> visitStatsDOIPage = visitStatsMapper.selectPage(
pageInfo,
new LambdaQueryWrapper<VisitStatsDO>()
.eq(VisitStatsDO::getProductId, request.getProductId())
);
/**
* 分页记录列表
*/
dataMap.put("current_data",visitStatsDOIPage.getRecords());
/**
* 当前分页总页数
*/
dataMap.put("total_page",visitStatsDOIPage.getPages());
/**
* 总记录数
*/
dataMap.put("total_record",visitStatsDOIPage.getTotal());
return dataMap;
}
}