从ClickHouse迁移到Doris集群：实战中如何搞定Array字段的平滑过渡（含Java代码示例）

在数据架构升级的浪潮中，许多团队正从ClickHouse转向Doris集群以应对更大规模的数据处理需求。这种迁移往往伴随着数据模型的重构，其中Array类型字段的适配尤为关键。本文将深入探讨两种数据库在Array处理上的核心差异，并提供一套完整的迁移方案，帮助开发者规避常见陷阱。

1. 理解两种数据库的Array实现差异

ClickHouse和Doris虽然都支持Array数据类型，但在底层实现和使用方式上存在显著不同。ClickHouse作为列式存储的先行者，其Array类型设计更偏向分析场景，支持复杂的嵌套操作。而Doris作为MPP架构的后起之秀，在保持高性能的同时，对标准SQL有更好的兼容性。

主要差异点对比：

特性	ClickHouse	Doris
存储模型	原生嵌套存储	序列化后存储
空值处理	区分 [] 和 NULL	统一视为 NULL
函数支持	提供 arrayJoin 等特有函数	遵循标准SQL规范
索引支持	支持数组元素索引	仅支持整体列索引
最大元素限制	无硬性限制	默认100万元素限制

在实际迁移中，最常遇到的挑战来自三个方面：

1. 数据序列化格式的转换
2. 查询语句中数组操作的语法差异
3. 性能优化策略的调整

2. 数据模型迁移策略

2.1 表结构设计最佳实践

Doris中创建包含Array字段的表时，需要考虑以下几个关键点：

CREATE TABLE IF NOT EXISTS traffic_metrics (
    `record_time` DATETIME NOT NULL COMMENT '记录时间',
    `intersection_id` INT NOT NULL COMMENT '路口ID',
    `phase_numbers` ARRAY<TINYINT> COMMENT '相位编号数组',
    `approach_metrics` ARRAY<STRING> COMMENT '进口指标数组'
)
DUPLICATE KEY(`record_time`, `intersection_id`)
DISTRIBUTED BY HASH(`intersection_id`) BUCKETS 8
PROPERTIES (
    "replication_num" = "3",
    "storage_medium" = "SSD"
);

注意：Doris 2.0+版本才支持Unique模型中的Array类型字段，在早期版本中只能在Duplicate模型中使用。

2.2 数据导入的三种方式

方式一：批量INSERT语句

INSERT INTO traffic_metrics VALUES
('2023-01-01 08:00:00', 1001, [1,2,3,4], ['N-0.85-0.12', 'S-0.78-0.15']),
('2023-01-01 08:05:00', 1001, [1,3], ['N-0.82-0.11', 'S-0.80-0.14']);

方式二：Stream Load导入

curl --location-trusted -u user:passwd \
-H "format: json" -H "strip_outer_array: true" \
-T data.json http://fe_host:8030/api/db/traffic_metrics/_stream_load

其中data.json文件内容示例：

[
    {
        "record_time": "2023-01-01 08:00:00",
        "intersection_id": 1001,
        "phase_numbers": [1,2,3,4],
        "approach_metrics": ["N-0.85-0.12", "S-0.78-0.15"]
    }
]

方式三：Java程序写入

public void insertTrafficData(TrafficData data) {
    String sql = "INSERT INTO traffic_metrics VALUES(?, ?, ?, ?)";
    
    try (Connection conn = dataSource.getConnection();
         PreparedStatement stmt = conn.prepareStatement(sql)) {
        
        // 处理普通字段
        stmt.setTimestamp(1, new Timestamp(data.getRecordTime().getTime()));
        stmt.setInt(2, data.getIntersectionId());
        
        // 处理Array字段 - 方案1：直接传递数组
        stmt.setArray(3, conn.createArrayOf("tinyint", data.getPhaseNumbers()));
        
        // 处理Array字段 - 方案2：拼接JSON字符串（兼容性更好）
        String approachStr = Arrays.stream(data.getApproachMetrics())
                                 .map(s -> "\"" + s.replace("\"", "\\\"") + "\"")
                                 .collect(Collectors.joining(",", "[", "]"));
        stmt.setString(4, approachStr);
        
        stmt.executeUpdate();
    }
}

3. 查询转换与函数映射

3.1 基础查询转换

ClickHouse中的常见数组查询在Doris中的对应实现：

ClickHouse查询：

SELECT arrayJoin(phase_numbers) AS phase 
FROM traffic_metrics 
WHERE intersection_id = 1001

Doris等效查询：

SELECT explode(phase_numbers) AS phase 
FROM traffic_metrics 
WHERE intersection_id = 1001

提示：Doris的 explode 是表函数，必须配合LATERAL VIEW使用，完整写法如下：

SELECT t.record_time, e.phase
FROM traffic_metrics t
LATERAL VIEW explode(t.phase_numbers) e AS phase
WHERE t.intersection_id = 1001

3.2 高级数组操作对照

1. 数组元素过滤

ClickHouse:

SELECT arrayFilter(x -> x > 2, phase_numbers) 
FROM traffic_metrics

Doris:

SELECT array_filter(phase_numbers, x -> x > 2)
FROM traffic_metrics

2. 数组聚合计算

ClickHouse:

SELECT arraySum(phase_numbers) AS total_phases
FROM traffic_metrics

Doris:

SELECT array_sum(phase_numbers) AS total_phases
FROM traffic_metrics

3. 多数组操作

ClickHouse:

SELECT arrayMap(x -> x*2, phase_numbers) AS doubled_phases
FROM traffic_metrics

Doris:

SELECT array_map(phase_numbers, x -> x*2) AS doubled_phases
FROM traffic_metrics

4. 性能优化专项

4.1 分区与分桶策略

针对包含Array字段的表，建议采用以下优化策略：

CREATE TABLE optimized_traffic_metrics (
    `dt` DATE NOT NULL COMMENT '日期分区',
    `hour` TINYINT NOT NULL COMMENT '小时',
    `intersection_id` INT NOT NULL,
    `metrics` ARRAY<FLOAT>,
    -- 其他字段...
)
PARTITION BY RANGE(`dt`) (
    PARTITION p202301 VALUES LESS THAN ('2023-02-01'),
    PARTITION p202302 VALUES LESS THAN ('2023-03-01')
)
DISTRIBUTED BY HASH(`intersection_id`, `hour`) BUCKETS 12
PROPERTIES (
    "replication_num" = "3",
    "storage_medium" = "SSD",
    "storage_cooldown_time" = "7 days"
);

4.2 数组大小控制策略

Doris对单个Array的大小有限制（默认1MB），可通过以下方式调整：

1. 修改BE配置 ：

# 在be.conf中添加
max_array_size = 1000000
max_array_byte_size = 10485760  # 10MB

2. 应用层分块处理 ：

public void insertLargeArray(List<BigData> items) {
    // 每1000条数据分一个批次
    List<List<BigData>> batches = Lists.partition(items, 1000);
    
    batches.forEach(batch -> {
        String arrayJson = batch.stream()
                              .map(BigData::toJson)
                              .collect(Collectors.joining(",", "[", "]"));
        // 执行插入...
    });
}

4.3 物化视图加速查询

对于频繁分析的数组元素，可以创建物化视图预计算：

CREATE MATERIALIZED VIEW mv_phase_stats
DISTRIBUTED BY HASH(intersection_id)
REFRESH ASYNC
AS SELECT
    intersection_id,
    dt,
    array_size(phase_numbers) AS phase_count,
    array_sum(phase_numbers) AS total_phases,
    array_avg(phase_numbers) AS avg_phase
FROM traffic_metrics

5. 实战问题排查指南

问题1：数组格式解析失败

症状 ：导入数据时出现"Array type parse failure"错误。

解决方案 ：

• 检查JSON格式是否合法，特别是字符串转义
• 验证元素类型是否与定义一致
• 使用 try_array() 函数捕获异常：

SELECT try_array(parse_json(column)) FROM table

问题2：explode性能低下

优化方案 ：

• 添加前置过滤条件减少处理数据量
• 考虑使用物化视图预计算
• 调整 exec_mem_limit 参数：

SET exec_mem_limit = 8589934592;  -- 8GB

问题3：数组函数不兼容

替代方案 ：

• 对于Doris不支持的函数，可以组合基本函数实现
• 例如实现 arrayDistinct ：

SELECT array_agg(distinct_val)
FROM (
    SELECT distinct val AS distinct_val
    FROM t LATERAL VIEW explode(arr) e AS val
) tmp

在一次实际迁移项目中，我们遇到一个包含200万条记录的表，其中Array字段平均包含50个元素。通过采用分区策略和优化分桶数量，查询性能从最初的12秒降低到800毫秒。关键是在测试环境充分验证各种边界情况，特别是处理NULL值和空数组时的行为差异。