数美科技大数据平台：从Hadoop到ClickHouse的实时风控演进

1. 数美科技大数据平台演进概述数美科技作为国内领先的在线业务风控服务商其大数据平台经历了从传统批处理到实时交互式分析的完整演进过程。早期平台采用典型的Hadoop生态架构数据查询响应时间普遍超过24小时严重制约了业务决策效率。而当前平台已实现定义即可查的实时交互能力查询延迟降低至秒级数据规模突破数百TB级别。这个转变背后是三个核心突破点首先是存储引擎从HDFS迁移到ClickHouse列式数据库使压缩比提升5倍的同时查询性能提高10倍以上其次是计算框架引入Spark SQL实现批流统一处理告别了原先MapReduce的繁重开发模式最后是采用JSON作为统一数据交换格式简化了上下游系统集成复杂度。2. 技术架构深度解析2.1 存储层ClickHouse的极致优化在ClickHouse集群部署上数美采用双副本分片集群模式每个分片由6节点组成。针对风控场景的特殊需求对默认配置进行了多项关键调整-- 建表示例展示特殊配置 CREATE TABLE risk_events ( event_time DateTime64(3, Asia/Shanghai), user_id String CODEC(ZSTD(3)), device_fp String CODEC(ZSTD(5)), risk_score Float32, features Nested( name String, value Float32 ) ) ENGINE ReplicatedMergeTree() PARTITION BY toYYYYMM(event_time) ORDER BY (user_id, event_time) TTL event_time INTERVAL 180 DAY SETTINGS index_granularity 8192;关键配置说明ZSTD压缩算法针对不同字段设置差异化的压缩级别设备指纹等长文本采用更高压缩比TTL机制自动清理过期数据DateTime64精确到毫秒满足风控时序需求。实际运行中遇到的最大挑战是稀疏列存储问题。当某些特征字段空值率超过90%时默认存储方式会浪费大量空间。通过启用allow_suspicious_low_cardinality_types参数对枚举型字段使用LowCardinality类型使存储空间减少40%。2.2 计算层Spark SQL的批流统一数美在Spark集群部署上选择YARN资源管理模式采用动态资源分配策略spark-submit \ --master yarn \ --deploy-mode cluster \ --conf spark.dynamicAllocation.enabledtrue \ --conf spark.shuffle.service.enabledtrue \ --conf spark.dynamicAllocation.maxExecutors100 \ --conf spark.sql.adaptive.enabledtrue \ --executor-memory 16G \ --executor-cores 4 \ --class com.shumei.risk.AnalysisJob \ risk-analysis.jar特征工程中大量使用Spark SQL的窗口函数实现滑动时间窗统计val behaviorFeatures spark.sql( SELECT user_id, COUNT(*) OVER ( PARTITION BY user_id ORDER BY event_time RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW ) AS hour_actions, AVG(risk_score) OVER ( PARTITION BY device_fp ORDER BY event_time ROWS BETWEEN 100 PRECEDING AND CURRENT ROW ) AS device_avg_score FROM risk_events WHERE event_time NOW() - INTERVAL 7 DAY )2.3 数据交换JSON Schema治理采用JSON Schema规范数据格式使用ajv工具进行校验{ $schema: http://json-schema.org/draft-07/schema#, type: object, properties: { event_time: { type: string, format: date-time }, risk_score: { type: number, minimum: 0, maximum: 1 }, ip_info: { type: object, properties: { country: {type: string}, asn: {type: integer} } } }, required: [event_time, risk_score] }在Spark中通过自定义SerDe实现JSON高效解析val schema new ArrayType( StructType(Seq( StructField(feature_name, StringType), StructField(feature_value, FloatType) )) ) spark.read.schema(schema) .option(mode, FAILFAST) .json(/data/events/*.json)3. 性能优化实战3.1 ClickHouse查询加速技巧针对风控场景的典型查询模式我们设计了特殊的物化视图CREATE MATERIALIZED VIEW risk_stats_hourly ENGINE AggregatingMergeTree() PARTITION BY toYYYYMMDD(event_time) ORDER BY (user_id, feature_type) POPULATE AS SELECT user_id, featureType AS feature_type, toStartOfHour(event_time) AS hour_time, countState() AS event_count, avgState(risk_score) AS avg_score FROM risk_events GROUP BY user_id, featureType, hour_time;查询时使用最终模式FINAL保证数据一致性SELECT user_id, sumMerge(event_count) AS total_actions, avgMerge(avg_score) AS overall_risk FROM risk_stats_hourly FINAL WHERE hour_time NOW() - INTERVAL 24 HOUR GROUP BY user_id HAVING total_actions 10 ORDER BY overall_risk DESC LIMIT 1000;3.2 Spark资源调优经验通过分析历史任务得出最佳资源配置比例任务类型Executor数量单Executor内存并行度系数特征计算5016G0.8模型训练3032G0.5数据清洗808G1.2关键Spark参数设置经验spark.sql.shuffle.partitions设为集群核心数×并行度系数对于JOIN操作设置spark.sql.autoBroadcastJoinThreshold64MB启用spark.speculationtrue应对慢节点问题4. 平台运维关键指标4.1 集群监控体系ClickHouse集群核心监控项指标名称预警阈值采集频率ReplicatedQueueSize100010sMemoryUsage90%5sMergeSpeed10MB/s1mQueryDuration99分位5s实时使用PrometheusGranafa构建的监控看板包含查询QPS热力图内存使用趋势副本同步延迟慢查询TOP104.2 数据质量检查每日例行数据质量检查项SELECT toDate(event_time) AS day, countIf(JSONIsValid(raw_data)0) AS invalid_json, countIf(user_id) AS empty_user, count() AS total, invalid_json/total AS bad_rate FROM risk_events WHERE event_time TODAY() - 1 GROUP BY day HAVING bad_rate 0.0001;5. 典型问题排查实录5.1 ClickHouse内存溢出现象查询报错Memory limit exceeded但实际数据量不大 根因GROUP BY字段基数过高导致中间状态爆炸 解决方案增加max_memory_usage参数对高基数字段先做采样或分桶使用approx_percentile替代精确计算5.2 Spark数据倾斜特征计算任务中某个stage耗时异常通过Spark UI定位到处理device_fp的task处理数据量是其他task的100倍解决方案val skewedKeys Seq(device123,device456) val broadcastKeys spark.sparkContext.broadcast(skewedKeys) df.withColumn(is_skewed, when(col(device_fp).isin(broadcastKeys.value), true).otherwise(false)) .repartition(100, col(is_skewed), rand())6. 平台演进路线当前架构仍面临两个主要挑战首先是实时特征计算延迟需要从当前的5分钟降低到30秒以内计划引入Flink替换部分Spark Streaming作业其次是跨数据中心查询性能问题正在测试ClickHouse的分布式表引擎。下一步重点将围绕以下方向基于GPU加速的特征计算自适应查询优化自动化数据分级存储