1：项目背景

在日常的工作中，业务同学随时需要对比多种纬度的数据，快速的做出决策。(分析页面的漏斗、实时流量、曝光点击/转化) 所以我们需要沉淀一套实时的组件，降低数据使用门槛，提升决策效率，在一定方向上驱动业务增长。

2：Real Time Olap

联机分析处理Olap是一种软件技术，它使分析人员能够迅速、一致、交互地从各个方面观察信息，以达到深入理解数据的目的。它具有FASMI(Fast Analysis of Shared Multidimensional Information)，即共享多维信息的快速分析的特征。其中F是快速性(Fast)，指系统能在数秒内对用户的多数分析要求做出反应；A是可分析性(Analysis)，指用户无需编程就可以定义新的专门计算，将其作为分析的一部 分，并以用户所希望的方式给出报告；M是多维性(Multi—dimensional)，指提供对数据分析的多维视图和分析；I是信息性(Information)，指能及时获得信息，并且管理大容量信息。
OLAP 分类，详细如下：

3：why clickhouse

clickhouse 可以不依赖任何组件，支持分布存储，响应快。对内存的要求没那么高，并且有丰富的查询函数支持（漏斗函数，留存函数）。详细见：clickhouse.tech/docs/zh/sql… 最终选择clickhouse 支持实时报表分析。

4：实时分析应该怎么做

1：全局属性建设

clickhouse 对于多表的join支持不好，建议所有的埋点事件都维护在一个大宽表。如果app的埋点体系设计的比较合理，并且不存在同名属性存在不同类型。那么只需要维护一个全局的属性表，把clickhouse的数据类型与hive埋点的数据类型mapping起来即可。详细如下：

我们的bi系统对于所有的埋点元数据强一致性管理，历史存在同名属性，不同类型的数据做了历史数据处理。注意修改完数据类型需要执行一次insert overwrite 全表，不然load历史数据到clickhouse会存在数据类型不一致的情况。相关建表语句如下：

CREATE TABLE dbname.distributed_table_name on cluster clustername
(
    distinct_id String,
    event String,
    staticdate Date,
    project String,
    event_date_time Nullable(DateTime)
) ENGINE = Distributed('clustername', 'default', localtablename, rand());
复制代码

2：那些埋点数据写入clickhouse

我们的app历史埋点建设比较混乱，经历了N轮产品。大家对于埋点的理解都不统一，最简单的做法就是新增埋点。我们只能在新的埋点规范大家的体系建设，逐步替换老的埋点体系 。所以并不是所有的埋点都写入clickhouse，而是业务选择对应事件的时候，判断clickhouse是否已同步相关数据。如果没有则通过spark launcher同步历史数据写入clickhouse ,实时数据隔天写入。（系统初始化的时候，已经将常用埋点分析初始化。）如果大家在实际的工作中，需要发起spark任务，建议通过livy。（rest服务即可完成交互，详细介绍点我）

注意clickhouse写入数据的时候需要唯一的key,我们是通过用户id+event+event_time 组成唯一key，而且我们的app不允许多端登录。

 val frame: DataFrame = spark.sql(readSQL).na.fill("主键字段如果为空给默认值")
    frame.write.format("jdbc")
      .mode(SaveMode.Append)
      .option("driver", clickHouseProperties.getProperty("jdbc.driver"))
      .option("url", clickHouseProperties.getProperty("jdbc.url"))
      .option("dbtable", projectMappingMap(eventInfo.getProject).getCkSchema)
      .option("user", clickHouseProperties.getProperty("jdbc.username"))
      .option("password", clickHouseProperties.getProperty("jdbc.password"))
      .option("batchsize", "100000")
      .option("isolationLevel", "NONE")
      .option("numPartitions", partition)
      .save()
复制代码

3：flink实时数据写入

clickhouse 写入数据不支持事务。如果实时任务挂了以后，基本无法保障hive的数据与clickhouse的数据是一致的。我们的任务调度是dolphinscheduler，任务挂了以后加了保活机制。在调度重启之前把当天的历史数据删除，重置kafka消费时间节点，所有数据重新写入。

clickhouse stream 启动的时候先删除数据
clickhouse-client --port 9002 -u xxx --password xxx -h xxx.com --query "alter table tablename on cluster clustername delete where staticdate >= '${1}'"


${2}是动态传入的时间点
kafka-consumer-groups.sh --bootstrap-server xxx:9092 --group ClickHouseSyncStream --reset-offsets --topic event --execute --to-datetime ${2}T15:59:00.000
复制代码

flink 写入clickhouse 通过window 添加 trigger 触发相关sink 操作（注意我们在实际的生产过程中遇到clickhouse 写入失败，建议把 socket_timeout=600000设置大一些）

 eventStream.process(new EventFilterProcess())//过滤
                .connect(broadcast)
                .process(new EventParseProcess(startSyncTime, eventInfoStateDescriptor))
                .keyBy(new EventKeyBy(100))
               .window(TumblingProcessingTimeWindows.of(Time.seconds(10))).trigger(new CustomCountTriggerWithEventTime<>(3000))//触发器
复制代码

3：业务系统如何查询

        业务系统在设计的时候一定要让业务同学可以操作，并且支持过滤相关数据。日常的实时分析可以通过bi建立好相关场景，把相关漏斗、自定义报表分享给业务即可。clickhouse数据查询响应存在一定的时间差，所以建议前端请求通过轮训。业务侧在处理请求的时候把相关配置改成异步。springboot的 Async 配置，即可完成相关异步操作。系统只需要引入clickhouse的windowFunnel即可完成漏斗计算，自定义报表可以通过clickhouse的groupArray完成分组数据的展示。

@EnableAsync
public class ClickhouseServiceApplication {

}
复制代码

clickhouse 的jdbc连接直接通JdbcTemplate 引入 clickhouseJdbcTemplate，详细如下：

@Configuration
public class ClickhouseDataSourceConfig implements BeanFactoryPostProcessor, EnvironmentAware {
    private ConfigurableEnvironment environment;
    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
        ClickHouseProperties clickHouseProperties=resolverSetting();
        createDataSourceBean(configurableListableBeanFactory, clickHouseProperties);
    }

    public void createDataSourceBean(ConfigurableListableBeanFactory beanFactory,
                                     ClickHouseProperties sqlProperties) {

        DataSource baseDataSource = clickHouseDruidDatasource(sqlProperties);
        register(beanFactory, new JdbcTemplate(baseDataSource), "clickhouseJdbcTemplate",
                "clickhouse");
    }

    private void register(ConfigurableListableBeanFactory beanFactory, Object bean, String name,
                          String alias) {
        beanFactory.registerSingleton(name, bean);
        if (!beanFactory.containsSingleton(alias)) {
            beanFactory.registerAlias(name, alias);
        }
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = (ConfigurableEnvironment) environment;
    }

    private ClickHouseProperties resolverSetting() {

        Iterable<ConfigurationPropertySource> sources = ConfigurationPropertySources.get(environment);
        Binder binder = new Binder(sources);
        BindResult<ClickHouseProperties> bindResult = binder.bind("clickhouse", ClickHouseProperties.class);
        ClickHouseProperties clickHouseProperties= bindResult.get();
        return  clickHouseProperties;
    }

    public DruidDataSource clickHouseDruidDatasource(ClickHouseProperties clickHouseProperties) {
        DruidDataSource datasource = new DruidDataSource();
        datasource.setUrl(clickHouseProperties.getUrl());
        datasource.setUsername(clickHouseProperties.getUsername());
        datasource.setPassword(clickHouseProperties.getPassword());
        datasource.setDriverClassName(clickHouseProperties.getDriverClassName());
        // pool configuration
        datasource.setInitialSize(clickHouseProperties.getInitialSize());
        datasource.setMinIdle(clickHouseProperties.getMinIdle());
        datasource.setMaxActive(clickHouseProperties.getMaxActive());

        datasource.setMaxWait(clickHouseProperties.getMaxWait());
        datasource.setTimeBetweenEvictionRunsMillis(clickHouseProperties.getTimeBetweenEvictionRunsMillis());
        datasource.setMinEvictableIdleTimeMillis(clickHouseProperties.getMinEvictableIdleTimeMillis());
        datasource.setValidationQuery(clickHouseProperties.getValidationQuery());
        return datasource;
    }
}
复制代码

4：总结

clickhouse 写入性能极高，并且占用的内存以及cpu相对比较合理。如果公司业务在百万级别，并且机器资源并不是很宽松，建议使用clickhouse,即可完成相关业务的支撑。当然像impala+kudu很多大型互联公司使用的比较多，在我看来适合自己的就是最好的。

​