I am trying to understand how to choose the shard key in Clickhosue ? and how clickhosue chooses which shard? for example, i have a table with 3 columns : user_id, timestamp, city_id.
should i shard by user_id or by City?
i use murmurHash3_64 function.
murmurHash3_64(city_id = 1) return :
First of all you need to understand why do you need sharding by some meaningful column. Why you could not use rand()?
Usually it's not a question what to use, because the sharding naturally follows the business requirements. If you don't have such requirements then you should use rand().
CH uses modulo operation + weight . It's very simple. If you have 6 shards then 956517343494314387 % 6 = 5 === shard number 5. So the rows with the same city_id will be placed on the same shard.
So if you chose city_id as shard key and the distribution by the city usually unequal so the shading will be unequal too. All rows for the big cities like New York or Mexico will in the one shard.
So user_id looks more appropriate as sharding key.
必须指定索引列，ClickHouse 中的**索引列即排序列，通过 order by 指定，一般在查询条件中经常被用来充当筛选条件的属性被纳入进来；可以是单一维度，也可以是组合维度的索引；通常需要满足高级列在前、查询频率大的在前原则；还有基数特别大的不适合做索引列，如用户表的 userid 字段；通常筛选后的数据满足在百万以内为最佳。**
echo "
select *
from realtime.kw_search_play_mixxxxxid_newformula_hour_approx_h
where etldate in ('2021-11-01')
and hour='17'
and mixxxxxid = 'xxxxxxx' and kw = 'xx' format TabSeparatedWithNames
" | curl 'http://xx.x.xxx.xx:8123/?user=xx_xxxxx&password=xx_xxxxx' -d @- > "${dst_path}/${filename_txt}"
check_file_size "${dst_path}/${filename_txt}"
SELECT 0 / 0
┌─divide(0, 0)─┐
│ nan │
└──────────────┘
case when isNaN(play_last_new) then 0 when isInfinite(play_last_new) then 0 else play_last_new end as play_last_new,
logfile=$BIPROG_ROOT/logs/`basename $0`.log #定义写日志文件名
###############################引入公共函数库#########################
source $BIPROG_ROOT/bin/shell/common.sh
###定义内部变量##################
YYYYMMDD=`echo ${DATA_DATE}|cut -c1-8`
vHour=`echo ${DATA_DATE}|cut -c9-10`
vDay=`date -d "$YYYYMMDD" +"%Y-%m-%d"`
vMonthExpired=`date -d "$YYYYMMDD 40 days ago" +"%Y%m"`
vRealyHour=${vDay}" "${vHour}
v1Hourbefore=`date -d "-1 hour $vRealyHour" +"%Y-%m-%d %H"`
v3Hourbefore=`date -d "-3 hour $vRealyHour" +"%Y-%m-%d %H"`
v3Hourbefore180=`date -d "-180 days -3 hour $vRealyHour" +"%Y-%m-%d %H"`
v6Hourbefore=`date -d "-6 hour $vRealyHour" +"%Y-%m-%d %H"`
v24Hourbefore=`date -d "-24 hour $vRealyHour" +"%Y-%m-%d %H"`
[ -z "$vDay" ] && { echo "no vDay founded."; exit 1; }
vDayYYYYMMDD=`echo ${vDay}|tr -d '-'`
vDay8daysago=`date -d "$vDay +7 days ago" +%Y%m%d`
writelog "vDay8daysago=${vDay8daysago}"
mr_queue=root.tenx
############################ 功能执行部分 ############################
writelog "Program($VERSION) start..."
writelog ${vDay}
writelog ${v24Hourbefore}
writelog ${vRealyHour}
echo "save day is: $vDay"
echo "save hour is: $vHour"
echo "right now is: $vRealyHour"
echo "one hour before is: $v1Hourbefore"
echo "24 hours befor is $v24Hourbefore"
efficient=XXXXXXXXXXXXXXXXXXXXXXXX
echo "insert into realtime.kw_search_play_mixxxxxid_newsong_h"
echo "insert into realtime.kw_search_play_mixxxxxid_newsong_h
select
etldate
, hour
, kw
, mixxxxxid
, overplay*100/$efficient
, search_play*100/$efficient
, search_play_30*100/$efficient
, click_cnt*100/$efficient
, kw_click_cnt
, round(play_last*100/$efficient, 6) as play_last
, click_cnt_all*100/$efficient
, kw_click_cnt_all
, round(play_last_all*100/$efficient, 6) as play_last_all
, round(play_last_xxxxxson*100/$efficient, 6) as play_last_xxxxxson
, round(play_last_all_xxxxxson*100/$efficient, 6) as play_last_all_xxxxxson
from
(
select
etldate, hour, kw, mixxxxxid, overplay, search_play, search_play_30, click_cnt, kw_click_cnt, play_last, click_cnt_all
, kw_click_cnt_all, play_last_all, play_last_xxxxxson, play_last_all_xxxxxson, b.upload_time
from
(
select
'${vDay}' as etldate, '${vHour}' as hour, kw, mixxxxxid, overplay, search_play, search_play_30
, click_cnt, kw_click_cnt, play_last, click_cnt_all, kw_click_cnt_all, play_last_all, play_last_xxxxxson, play_last_all_xxxxxson
from
(
select
kw, a.mixxxxxid, overplay, search_play, search_play_30, click_cnt, kw_click_cnt
, case when b.mixxxxxid<>'' then play_last*X else play_last end as play_last
, click_cnt_all, kw_click_cnt_all
, case when b.mixxxxxid<>'' then play_last_all*X else play_last_all end as play_last_all
, case when b.mixxxxxid<>'' then play_last_xxxxxson*X else play_last_xxxxxson end as play_last_xxxxxson
, case when b.mixxxxxid<>'' then play_last_all_xxxxxson*X else play_last_all_xxxxxson end as play_last_all_xxxxxson
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, a.click_cnt, a.kw_click_cnt, a.play_last, play_last_xxxxxson
, a.click_cnt_all, b.kwplay as kw_click_cnt_all, a.play_last_all, play_last_all_xxxxxson
from
(
select
XXXXXXXXXXXXXXXXXXXXXXXX
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, a.click_cnt, a.kw_click_cnt, a.play_last, a.play_last_xxxxxson, b.search_play as click_cnt_all
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, a.click_cnt, b.kwplay as kw_click_cnt, a.play_last, a.play_last_xxxxxson
from
(
select
XXXXXXXXXXXXXXXXXXXXXXXX
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, b.search_play as click_cnt
from
(
select
kw, scid_albumid as mixxxxxid, sum(overplay) as overplay
, sum(search_play) as search_play, sum(search_play_30) as search_play_30
from realtime.kw_search_play_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v1Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
and kw<>''
and lower(kw) not like '%dj%'
and kw not like '%原唱%'
and kw not like '%免费%'
group by kw, scid_albumid
)a
left join
(
select kw, mixxxxxid, sum(search_play) as search_play
from realtime.kw_search_click_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v1Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw, mixxxxxid
)b using mixxxxxid, kw
)a
left join
(
select mixxxxxid
from dal.mixxxxxid_timelength_30_d
where etldate =subtractDays(toDate('${vDay}'),2)
)b using mixxxxxid
)a
left join
(
select kw, sum(search_play) as kwplay
from realtime.kw_search_click_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v1Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw
)b using kw
)a
left join
(
select kw, mixxxxxid, sum(search_play) as search_play
from realtime.kw_search_click_all_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v1Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw, mixxxxxid
)b using mixxxxxid, kw
)a
left join
(
select mixxxxxid
from dal.mixxxxxid_timelength_30_d
where etldate =subtractDays(toDate('${vDay}'),2)
)b using mixxxxxid
)a
left join
(
select kw, sum(search_play) as kwplay
from realtime.kw_search_click_all_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v1Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw
)b using kw
)a
left join
(
select mixxxxxid
from dal.dq_xxxxx_song
)b using mixxxxxid
)t where play_last<>0 or play_last_all<>0
)a
left join
(
select
mixxxxxid, max(upload_time) as upload_time
from (
select mixxxxxid, upload_time
from dal.oa_mixsong_new_d
where is_publish='1'
and substring(upload_time, 1, 13)>='${v24Hourbefore}'
and substring(upload_time, 1, 13)<'${vRealyHour}'
group by mixxxxxid, upload_time
)a group by mixxxxxid
)b using mixxxxxid
)t where upload_time<>''
" | POST 'http://xx.x.xxx.xx:8123/?user=huge_insert&password=huge_insert@xxxxx' | tee /tmp/xx_tmp_$$.log
[ -s /tmp/xx_tmp_$$.log ] && { rm -f /tmp/xx_tmp_$$.log; exit 1; }
rm -f /tmp/xx_tmp_$$.log
echo sleep 180
sleep 180
echo "insert into realtime.kw_search_play_mixxxxxid_combinesong_h"
echo "insert into realtime.kw_search_play_mixxxxxid_combinesong_h
select
etldate
, hour
, kw
, mixxxxxid
, overplay
, search_play
, search_play_30
, click_cnt
, kw_click_cnt
, play_last
, click_cnt_all
, kw_click_cnt_all
, play_last_all
, play_last_xxxxxson
, play_last_all_xxxxxson
from
(
select
etldate
, hour
, kw
, mixxxxxid
, overplay
, search_play
, search_play_30
, click_cnt
, kw_click_cnt
, play_last
, click_cnt_all
, kw_click_cnt_all
, play_last_all
, play_last_xxxxxson
, play_last_all_xxxxxson
, b.upload_time
from
(
select
'${vDay}' as etldate, '${vHour}' as hour, kw, mixxxxxid, overplay, search_play, search_play_30
, click_cnt, kw_click_cnt, play_last, click_cnt_all, kw_click_cnt_all, play_last_all, play_last_xxxxxson, play_last_all_xxxxxson
from
(
select
kw, a.mixxxxxid, overplay, search_play, search_play_30, click_cnt, kw_click_cnt
, case when b.mixxxxxid<>'' then play_last*X else play_last end as play_last
, click_cnt_all, kw_click_cnt_all
, case when b.mixxxxxid<>'' then play_last_all*X else play_last_all end as play_last_all
, case when b.mixxxxxid<>'' then play_last_xxxxxson*X else play_last_xxxxxson end as play_last_xxxxxson
, case when b.mixxxxxid<>'' then play_last_all_xxxxxson*X else play_last_all_xxxxxson end as play_last_all_xxxxxson
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, a.click_cnt, a.kw_click_cnt, a.play_last, play_last_xxxxxson
, a.click_cnt_all, b.kwplay as kw_click_cnt_all, a.play_last_all, play_last_all_xxxxxson
from
(
select
XXXXXXXXXXXXXXXXXXXXXXXX
from
(
select
a.kw, a.mixxxxxid, a.overplay, a.search_play, a.search_play_30
, a.click_cnt, a.click_cnt_all, b.kwplay as kw_click_cnt, a.play_last, a.play_last_xxxxxson
from
(
select
XXXXXXXXXXXXXXXXXXXXXXXX
from
(
select kw, mixxxxxid, avg(overplay) as overplay, avg(search_play) as search_play, avg(search_play_30) as search_play_30, avg(click_cnt) as click_cnt, avg(click_cnt_all) as click_cnt_all
from (
select etldate, hour, kw, mixxxxxid, overplay, search_play, search_play_30, click_cnt, click_cnt_all
from realtime.kw_search_play_mixxxxxid_newsong_h
where concat(toString(etldate),' ',hour) >='${v24Hourbefore}'
and concat(toString(etldate),' ',hour) <'${vRealyHour}'
)a group by kw, mixxxxxid
)a
left join
(
select mixxxxxid
from dal.mixxxxxid_timelength_30_d
where etldate =subtractDays(toDate('${vDay}'),2)
)b using mixxxxxid
)a
left join
(
select kw, sum(search_play) as kwplay
from realtime.kw_search_click_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v24Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw
)b using kw
)a
left join
(
select mixxxxxid
from dal.mixxxxxid_timelength_30_d
where etldate =subtractDays(toDate('${vDay}'),2)
)b using mixxxxxid
)a
left join
(
select kw, sum(search_play) as kwplay
from realtime.kw_search_click_all_h
where dt between subtractDays(toDate('${vDay}'),1) and '${vDay}'
and concat(toString(dt),' ',hour) >='${v24Hourbefore}'
and concat(toString(dt),' ',hour) <'${vRealyHour}'
group by kw
)b using kw
)a
left join
(
select mixxxxxid
from dal.dq_xxxxx_song
)b using mixxxxxid
)t where play_last<>0 or play_last_all<>0
)a
left join
(
select
mixxxxxid, max(upload_time) as upload_time
from (
select mixxxxxid, upload_time
from dal.oa_mixsong_new_d
where is_publish='1'
and substring(upload_time, 1, 13)>='${v24Hourbefore}'
and substring(upload_time, 1, 13)<'${vRealyHour}'
group by mixxxxxid, upload_time
)a group by mixxxxxid
)b using mixxxxxid
)t where upload_time<>''
" | POST 'http://xx.x.xxx.xx:8123/?user=huge_insert&password=huge_insert@xxxxx' | tee /tmp/xx_tmp_$$.log
[ -s /tmp/xx_tmp_$$.log ] && { rm -f /tmp/xx_tmp_$$.log; exit 1; }
rm -f /tmp/xx_tmp_$$.log
echo sleep 180
sleep 180
# # 清理
# if [ `echo ${DATA_DATE}|cut -c7-8` -eq 15 ]
# then
# echo "alter table realtime_local.kw_search_play_mixxxxxid_newsong_h ON CLUSTER xx_xx_cluster drop partition ${vMonthExpired}" | POST 'http://xx.x.xxx.xx:8123/?user=insert&password=insert@xxxxx' | tee /tmp/xx_tmp_$$.log
# rc=`grep -c "DB::Exception" /tmp/xx_tmp_$$.log`
# rm -f /tmp/xx_tmp_$$.log
# if [ `echo ${DATA_DATE}|cut -c7-8` -eq 15 ]
# then
# echo "alter table realtime_local.kw_search_play_mixxxxxid_combinesong_h ON CLUSTER xx_xx_cluster drop partition ${vMonthExpired}" | POST 'http://xx.x.xxx.xx:8123/?user=insert&password=insert@xxxxx' | tee /tmp/xx_tmp_$$.log
# rc=`grep -c "DB::Exception" /tmp/xx_tmp_$$.log`
# rm -f /tmp/xx_tmp_$$.log
writelog "Program finish successfully."
SET max_memory_usage = 128000000000; #128G，
set max_bytes_before_external_group_by=20000000000; #20G
set max_memory_usage=40000000000; #40G
:) SET max_memory_usage = 90000000000
Ok.
:) SET max_bytes_before_external_group_by = 30000000000
Ok.
:) select user_id,tool_id,sum(amount) from big_sgz_2017.bi_tool_list_all where dateline between '2021-03-23' and '2021-03-28' and amount > 0 and bi_pid in ('19','222','463','464','502','549','1015') group by user_id,tool_id
↘ Progress: 273.08 million rows, 15.29 GB (1.66 million rows/s., 93.05 MB/s.) ███████████████████████████████████████████████████████████████████████████████▋ 73%
Exception on client:
Code: 32. DB::Exception: Attempt to read after eof: while receiving packet from localhost:9000
Connecting to database prod at localhost:9000 as user default.
Code: 210. DB::NetException: Connection refused (localhost:9000)
