ClickHouse：人群圈选业务的大杀器 - 今日头条

1

ALTER TABLE [db.]table UPDATE column1 = expr1 [, ...] WHERE filter_expr;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

CREATE TABLE IF NOT EXISTS whatever_table ON CLUSTER default
(
    user_id UInt64,
    city_level SimpleAggregateFunction(anyLast, Nullable(Enum('一线城市' = 0, '二线城市' = 1, '三线城市' = 2, '四线城市' = 3))),
    gender SimpleAggregateFunction(anyLast, Nullable(Enum('女' = 0, '男' = 1))),
    interest_sports SimpleAggregateFunction(anyLast, Nullable(Enum('否' = 0, '是' = 1))),
    reg_date SimpleAggregateFunction(anyLast, Datetime),
    comment_like_cnt SimpleAggregateFunction(anyLast, Nullable(UInt32)),
    last30d_share_cnt SimpleAggregateFunction(anyLast, Nullable(UInt32)),
    user_like_consume_trend_type SimpleAggregateFunction(anyLast, Nullable(String)),
    province SimpleAggregateFunction(anyLast, Nullable(String)),
    last_access_version SimpleAggregateFunction(anyLast, Nullable(String)),
    others SimpleAggregateFunction(anyLast,Array(String))
)ENGINE = AggregatingMergeTree() partition by toYYYYMMDD(reg_date) ORDER BY user_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
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

import java.sql.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.TimeoutException;




public class Main {
    private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    private static final SimpleDateFormat SIMPLE_DATE_FORMAT = new SimpleDateFormat(DATE_FORMAT);


    public static void main(String[] args) throws ClassNotFoundException, SQLException, InterruptedException, ParseException {
        String url = "your url";
        String username = "your username";
        String password = "your password";


        Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
        String connectionStr = "jdbc:clickhouse://" + url + ":8123";


        try {
            Connection connection = DriverManager.getConnection(connectionStr, username, password);
            Statement stmt = connection.createStatement();


            // 创建local表
            String createLocalTableDDL = "CREATE TABLE IF NOT EXISTS whatever_table ON CLUSTER default " +
                    "(user_id UInt64, " +
                    "city_level SimpleAggregateFunction(anyLast, Nullable(Enum('一线城市' = 0, '二线城市' = 1, '三线城市' = 2, '四线城市' = 3))), " +
                    "gender SimpleAggregateFunction(anyLast, Nullable(Enum('女' = 0, '男' = 1)))," +
                    "interest_sports SimpleAggregateFunction(anyLast, Nullable(Enum('否' = 0, '是' = 1)))," +
                    "reg_date SimpleAggregateFunction(anyLast, Datetime)) " +
                    "comment_like_cnt SimpleAggregateFunction(anyLast, Nullable(UInt32)),\n" +
                    "last30d_share_cnt SimpleAggregateFunction(anyLast, Nullable(UInt32)),\n" +
                    "user_like_consume_trend_type SimpleAggregateFunction(anyLast, Nullable(String)),\n" +
                    "province SimpleAggregateFunction(anyLast, Nullable(String)),\n" +
                    "last_access_version SimpleAggregateFunction(anyLast, Nullable(String)),\n" +
                    "others SimpleAggregateFunction(anyLast, Array(String)),\n" +
                    "ENGINE = AggregatingMergeTree() PARTITION by toYYYYMM(reg_date) ORDER BY user_id;";


            stmt.execute(createLocalTableDDL);
            System.out.println("create local table done.");


            // 创建distributed表
            String createDistributedTableDDL = "CREATE TABLE IF NOT EXISTS whatever_table_dist ON cluster default " +
                    "AS default.whatever_table " +
                    "ENGINE = Distributed(default, default, whatever_table, intHash64(user_id));";
            stmt.execute(createDistributedTableDDL);
            System.out.println("create distributed table done");


            // 插入mock数据
            String insertSQL = "INSERT INTO whatever_table(\n" +
                    "\tuser_id,\n" +
                    "\tcity_level,\n" +
                    "\tgender,\n" +
                    "\tinterest_sports,\n" +
                    "\treg_date,\n" +
                    "\tcomment_like_cnt,\n" +
                    "\tlast30d_share_cnt,\n" +
                    "\tuser_like_consume_trend_type,\n" +
                    "\tprovince,\n" +
                    "\tlast_access_version,\n" +
                    "\tothers\n" +
                    "\t)SELECT\n" +
                    " number as user_id,\n" +
                    " toUInt32(rand(11)%4) as city_level,\n" +
                    " toUInt32(rand(30)%2) as gender,\n" +
                    " toUInt32(rand(28)%2) as interest_sports,\n" +
                    " (toDateTime('2020-01-01 00:00:00') + rand(1)%(3600*24*30*4)) as reg_date,\n" +
                    " toUInt32(rand(15)%10) as comment_like_cnt,\n" +
                    " toUInt32(rand(16)%10) as last30d_share_cnt,\n" +
                    "randomPrintableASCII(64) as user_like_consume_trend_type,\n" +
                    "randomPrintableASCII(64) as province,\n" +
                    "randomPrintableASCII(64) as last_access_version,\n" +
                    "[randomPrintableASCII(64)] as others\n" +
                    " FROM numbers(100000);\n";
            stmt.execute(insertSQL);
            System.out.println("Mock data and insert done.");


            System.out.println("Select count(user_id)...");
            ResultSet rs = stmt.executeQuery("select count(user_id) from whatever_table_dist");
            while (rs.next()) {
                int count = rs.getInt(1);
                System.out.println("user_id count: " + count);
            }


            // 数据合并
            String optimizeSQL = "OPTIMIZE table whatever_table final;";
            // 如数据合并时间过长，可在partition级别并行执行
            String optimizeByPartitionSQL = "OPTIMIZE table whatever_table PARTITION 202001 final;";
            try {
                stmt.execute(optimizeByPartitionSQL);
            }catch (SQLTimeoutException e){
                // 查看merge进展
                // String checkMergeSQL = "select * from system.merges where database = 'default' and table = 'whatever_table';";
                Thread.sleep(60*1000);
            }




            // 人群圈选（city_level='一线城市'，gender='男性'，interest_sports='是', reg_date<='2020-01-31 23:59:59'）
            String selectSQL = "SELECT user_id from whatever_table_dist where city_level=0 and gender=1 and interest_sports=1 and reg_date <= NOW();";
            rs = stmt.executeQuery(selectSQL);
            while (rs.next()) {
                int user_id = rs.getInt(1);
                System.out.println("Got suitable user: " + user_id);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}