中证数据从交易所、中国结算、客户系统获取多源异构的指数数据（如行情、成交、持仓数据），这些数据格式和更新频率不同，请设计数据清洗与整合流程，并说明如何保证数据一致性？

1) 【一句话结论】：针对多源异构指数数据，需构建分层清洗与整合流程，通过统一数据模型、ETL/ELT流程、数据质量监控和版本控制，确保数据一致性。

2) 【原理/概念讲解】：首先，多源异构数据的核心挑战是“格式多样（如JSON、CSV、数据库表）”“更新频率不一（实时流、定时批量）”“数据源差异（交易所、中国结算、客户系统业务逻辑不同）”。数据清洗需解决格式转换、时间对齐、缺失/异常值处理；数据整合需通过ETL/ELT流程将多源数据加载到统一数据仓库，并使用CDC捕获增量更新。保证一致性的关键在于：① 统一数据模型（如星型模式，事实表+维度表，字段命名和类型统一）；② 数据校验规则（如业务逻辑校验，持仓量非负、价格非负）；③ 时间戳标准化（如统一为ISO 8601格式，秒级对齐）；④ 版本控制（如Git管理ETL脚本，跟踪数据流程变更）。

类比：可以把多源数据比作不同“语言”的人（格式、频率不同），清洗与整合流程就是“翻译”和“统一语言”的过程，最终让所有数据能“交流”并保持“一致”。

3) 【对比与适用场景】：

对比维度	ETL（Extract-Transform-Load）	ELT（Extract-Load-Transform）	适用场景	注意点
定义	先转换再加载	先加载再转换	数据量小、转换复杂度高（如复杂计算）	需要更多存储资源，转换逻辑复杂时效率低
特性	转换在加载前完成	转换在加载后完成	数据仓库（如星型模式）	数据湖（如HDFS+Spark）
使用场景	传统数据仓库（如Oracle、SQL Server）	大数据环境（如Hadoop、Spark）	交易所实时行情数据（需实时计算指标）	客户系统API数据（结构简单，转换少）
注意点	转换逻辑复杂易出错	加载后转换灵活，但需更多存储	需要稳定的数据源和转换规则	需要强大的计算资源

4) 【示例】：
假设从三个源获取数据：

• 交易所：实时流数据（JSON格式），字段：symbol（标的代码）、price（当前价）、volume（成交量）、timestamp（时间戳，毫秒级）；
• 中国结算：定时批量数据（CSV格式），字段：account（账户）、symbol（标的代码）、position（持仓量）、update_time（更新时间，字符串格式）；
• 客户系统：API接口（REST响应），字段：user_id（用户ID）、symbol（标的代码）、custom_metric（自定义指标）、timestamp（时间戳，ISO 8601格式）。

清洗与整合流程伪代码：

# 1. 时间戳标准化
def standardize_timestamp(data, source):
    if source == "exchange":
        return data["timestamp"] / 1000  # 毫秒转秒
    elif source == "csac":
        return datetime.strptime(data["update_time"], "%Y-%m-%d %H:%M:%S").timestamp()
    elif source == "client":
        return datetime.fromisoformat(data["timestamp"]).timestamp()

# 2. 格式转换与字段映射
def transform_data(data, source):
    if source == "exchange":
        return {
            "symbol": data["symbol"],
            "price": float(data["price"]),
            "volume": int(data["volume"]),
            "timestamp": standardize_timestamp(data, "exchange")
        }
    elif source == "csac":
        return {
            "account": data["account"],
            "symbol": data["symbol"],
            "position": int(data["position"]),
            "update_time": standardize_timestamp(data, "csac")
        }
    elif source == "client":
        return {
            "user_id": data["user_id"],
            "symbol": data["symbol"],
            "custom_metric": float(data["custom_metric"]),
            "timestamp": standardize_timestamp(data, "client")
        }

# 3. 缺失值处理
def handle_missing(data):
    for key in ["price", "volume", "position"]:
        if data[key] is None or data[key] == "":
            data[key] = data.get(f"prev_{key}", 0)  # 用前值填充

# 4. 加载到统一数据模型（数据仓库）
def load_to_warehouse(data, source):
    if source == "exchange":
        # 加载到行情事实表
        insert_into_facts("market_data", data)
    elif source == "csac":
        # 加载到持仓事实表
        insert_into_facts("position_data", data)
    elif source == "client":
        # 加载到自定义指标事实表
        insert_into_facts("custom_metrics", data)

# 主流程
def integrate_data():
    # 从各源抽取数据（假设通过消息队列或API获取）
    exchange_data = get_data_from_exchange()
    csac_data = get_data_from_csac()
    client_data = get_data_from_client()

    # 清洗与转换
    exchange_cleaned = [transform_data(d, "exchange") for d in exchange_data]
    csac_cleaned = [transform_data(d, "csac") for d in csac_data]
    client_cleaned = [transform_data(d, "client") for d in client_data]

    # 处理缺失值
    for d in exchange_cleaned:
        handle_missing(d)
    for d in csac_cleaned:
        handle_missing(d)
    for d in client_cleaned:
        handle_missing(d)

    # 加载到数据仓库
    load_to_warehouse(exchange_cleaned, "exchange")
    load_to_warehouse(csac_cleaned, "csac")
    load_to_warehouse(client_cleaned, "client")

# 启动整合流程
integrate_data()

5) 【面试口播版答案】：
面试官您好，针对中证数据多源异构指数数据的清洗与整合，我的设计思路是构建分层流程，核心是通过统一数据模型、ETL/ELT流程、数据质量监控和版本控制来保证一致性。首先，数据清洗阶段，针对不同源的数据格式（如JSON、CSV）和更新频率（实时流、定时批量），先进行格式转换（比如将CSV转为JSON，统一字段命名），然后处理时间戳对齐（比如将所有时间字段转换为统一格式并按秒级对齐），接着处理缺失值和异常值（比如用前值填充缺失价格，用阈值检测异常成交量）。然后整合阶段，采用ETL流程，从各源系统抽取数据，通过转换规则（如业务逻辑校验持仓量非负）加载到统一的数据仓库中，同时使用CDC（变更数据捕获）技术跟踪增量更新，确保数据一致性。最后通过数据质量监控（如每日校验规则）和版本控制（如Git管理ETL脚本）持续保障一致性。

6) 【追问清单】：

1. 如何处理不同数据源的更新频率差异？
   回答要点：通过CDC技术捕获增量更新，结合定时任务处理批量数据，实时流数据用消息队列缓冲。
2. 如果数据源存在数据冲突（比如同一时间点不同源的价格不一致），如何解决？
   回答要点：引入数据冲突解决机制，比如优先级规则（交易所数据优先于其他源），或通过人工审核标记冲突数据。
3. 数据清洗中的异常值检测如何实现？
   回答要点：结合规则引擎（如价格突变超过5%标记异常）和机器学习模型（如基于历史数据的异常检测算法）。
4. 如何保证数据加载后的一致性？
   回答要点：通过数据仓库的校验规则（如事实表的总计字段与明细数据一致）和定期全量校验。
5. 如果数据源发生变更（比如新增字段），如何快速适应？
   回答要点：使用配置管理（如YAML配置文件定义字段映射），通过自动化测试验证新流程。

7) 【常见坑/雷区】：

1. 忽略数据源更新频率差异，导致整合流程效率低或数据延迟；
2. 未考虑数据冲突解决机制，导致不一致；
3. 缺乏数据质量监控，无法及时发现数据问题；
4. 未使用统一数据模型，导致整合后数据不一致；
5. 忽略数据版本控制，导致ETL脚本变更后无法回溯。