3. 数据源探查与接入：MySQL、Hive、Kafka 等常见数据源接入方式与探查 SQL 示例

在风控标签口径校准项目中，数据探查是第一步，也是最容易被忽视的一步。不同数据源（OLTP、OLAP、消息队列）的接入方式、探查语法、性能特征差异巨大。本章将系统梳理 MySQL、Hive、Kafka 三种典型数据源的接入策略与探查 SQL 模板，帮助你在项目初期快速摸清数据“家底”。

3.1 数据源接入方式对比

不同数据源在风控架构中的角色不同，接入方式也各有侧重。下表总结了三种常见数据源的接入场景、驱动方式与探查工具。

3.2 MySQL 数据探查 SQL 示例

MySQL 探查的核心目标是：快速了解表结构、数据量、空值率、枚举字段分布。以下 SQL 模板覆盖了风控标签口径校准中最常用的探查场景。

-- 1. 表结构与行数概览
DESC risk_order_label;
SELECT COUNT(*) AS total_rows FROM risk_order_label;

-- 2. 空值率探查（关键字段）
SELECT
    COUNT(*) AS total,
    SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) AS user_id_null,
    SUM(CASE WHEN order_amount IS NULL THEN 1 ELSE 0 END) AS amount_null,
    ROUND(SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) / COUNT(*), 4) AS user_id_null_rate
FROM risk_order_label;

-- 3. 枚举字段值分布
SELECT label_status, COUNT(*) AS cnt, ROUND(COUNT(*) / SUM(COUNT(*)) OVER(), 4) AS ratio
FROM risk_order_label
GROUP BY label_status
ORDER BY cnt DESC;

-- 4. 时间范围探查（用于校准时间窗口）
SELECT MIN(create_time) AS min_time, MAX(create_time) AS max_time,
       DATEDIFF(MAX(create_time), MIN(create_time)) AS day_span
FROM risk_order_label;

-- 5. 探查采样（避免全表扫描）
SELECT * FROM risk_order_label TABLESAMPLE SYSTEM(1) LIMIT 1000;

3.3 Hive 数据探查 SQL 示例

Hive 探查的核心挑战是：分区裁剪、小文件合并、数据倾斜。风控标签宽表通常按天分区，探查时必须指定分区范围，否则会触发全表扫描。

-- 1. 查看表分区结构
SHOW PARTITIONS dwd_risk_label_di;

-- 2. 探查指定分区的行数与空值率
SELECT
    COUNT(*) AS total,
    SUM(CASE WHEN label_score IS NULL THEN 1 ELSE 0 END) AS score_null,
    ROUND(SUM(CASE WHEN label_score IS NULL THEN 1 ELSE 0 END) / COUNT(*), 4) AS score_null_rate
FROM dwd_risk_label_di
WHERE dt = '2025-03-01';

-- 3. 枚举字段分布（使用 APPROX 近似统计，避免 OOM）
SELECT label_type, COUNT(*) AS cnt
FROM dwd_risk_label_di
WHERE dt = '2025-03-01'
GROUP BY label_type
ORDER BY cnt DESC;

-- 4. 探查数据倾斜（按 key 分组统计）
SELECT user_id, COUNT(*) AS cnt
FROM dwd_risk_label_di
WHERE dt = '2025-03-01'
GROUP BY user_id
HAVING cnt > 1000
ORDER BY cnt DESC
LIMIT 10;

-- 5. 探查小文件数量（影响探查性能）
SHOW TBLPROPERTIES dwd_risk_label_di ('numFiles');

3.4 Kafka 数据探查示例

Kafka 探查的核心目标是：确认 Topic 是否存在、分区数、消息格式（JSON/Avro/Protobuf）、关键字段是否缺失。由于 Kafka 是流式存储，探查方式与 MySQL/Hive 完全不同。

# 1. 查看 Topic 列表与分区数（命令行）
kafka-topics.sh --bootstrap-server localhost:9092 --list
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic risk_event_raw

# 2. 消费最近 10 条消息，探查消息体结构（JSON 格式）
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic risk_event_raw \
  --partition 0 \
  --offset latest \
  --max-messages 10 \
  --property print.key=true \
  --property print.timestamp=true

# 3. 探查消息 Key 分布（用于判断是否按 user_id 分区）
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic risk_event_raw \
  --group probe_group \
  --from-beginning \
  --max-messages 1000 \
  --property print.key=true \
  --property print.value=false \
  | awk '{print $1}' | sort | uniq -c | sort -nr | head -10

// 4. 使用 Java/Python 客户端探查 Avro 消息（代码片段）
// 探查目标：确认 schema 兼容性、字段类型
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "probe_group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("risk_event_raw"));
ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(5000));
for (ConsumerRecord<String, GenericRecord> record : records) {
    GenericRecord value = record.value();
    System.out.println("event_type: " + value.get("event_type"));
    System.out.println("user_id: " + value.get("user_id"));
    System.out.println("timestamp: " + value.get("timestamp"));
    break; // 只探查第一条
}

3.5 数据源探查流程总览

下图展示了从数据源接入到探查结果输出的完整流程，适用于风控标签口径校准项目的初始化阶段。

3.6 探查结果汇总表示例

探查完成后，建议将结果汇总到一张表中，便于后续标签口径校准决策。以下是一个典型的探查结果汇总模板。