告别分库分表的噩梦：Apache Doris 3.0 如何成为 AI 时代的实时数仓基石

# docker-compose.yml
# Apache Doris 3.0 单机开发环境 + 监控套件
# 说明：生产环境请使用 3FE+3BE 高可用部署

version: '3.8'

networks:
  doris-net:
    driver: bridge

services:
  # ============ FE 前端节点 ============
  doris-fe:
    image: apache/doris:3.0.6-fe-x86_64
    container_name: doris-fe
    hostname: doris-fe
    environment:
      - FE_SERVERS=fe1:doris-fe:9010
      - FE_ID=1
    ports:
      - "8030:8030"   # HTTP API / Web UI
      - "9030:9030"   # MySQL 协议端口
      - "9010:9010"   # FE 内部通信
    volumes:
      - doris-fe-data:/opt/apache-doris/fe/doris-meta
      - doris-fe-log:/opt/apache-doris/fe/log
    networks:
      - doris-net
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8030/api/bootstrap"]
      interval: 30s
      timeout: 10s
      retries: 5
    restart: unless-stopped

  # ============ BE 后端节点 ============
  doris-be:
    image: apache/doris:3.0.6-be-x86_64
    container_name: doris-be
    hostname: doris-be
    environment:
      - FE_SERVERS=fe1:doris-fe:9010
      - BE_ADDR=doris-be:9050
    ports:
      - "8040:8040"   # BE HTTP 端口（Tablet 管理）
      - "9050:9050"   # BE 心跳端口
    volumes:
      - doris-be-data:/opt/apache-doris/be/storage
      - doris-be-log:/opt/apache-doris/be/log
    networks:
      - doris-net
    depends_on:
      doris-fe:
        condition: service_healthy
    restart: unless-stopped

  # ============ Prometheus 监控 ============
  prometheus:
    image: prom/prometheus:v2.51.0
    container_name: doris-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    networks:
      - doris-net
    restart: unless-stopped

  # ============ Grafana 可视化 ============
  grafana:
    image: grafana/grafana:10.4.0
    container_name: doris-grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=doris123
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/dashboards:/var/lib/grafana/dashboards
      - ./grafana/provisioning:/etc/grafana/provisioning
    networks:
      - doris-net
    depends_on:
      - prometheus
    restart: unless-stopped

volumes:
  doris-fe-data:
  doris-fe-log:
  doris-be-data:
  doris-be-log:
  prometheus-data:
  grafana-data:

# prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'doris_fe'
    static_configs:
      - targets: ['doris-fe:8030']
    metrics_path: '/metrics'

  - job_name: 'doris_be'
    static_configs:
      - targets: ['doris-be:8040']
    metrics_path: '/metrics'

# 创建目录结构
mkdir -p doris-demo/{grafana/dashboards,grafana/provisioning/datasources}
cd doris-demo

# 复制上面两个配置文件后启动
docker compose up -d

# 等待约 60 秒，检查状态
docker compose ps

# 验证 FE 存活
mysql -uroot -P9030 -h127.0.0.1 \
  -e "SELECT host, join, alive FROM frontends()"

# 验证 BE 存活
mysql -uroot -P9030 -h127.0.0.1 \
  -e "SELECT host, alive FROM backends()"

<dependencies>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- Spring Data JPA / JDBC -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <!-- MySQL Connector（Doris 兼容 MySQL 协议）-->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
    <!-- Thymeleaf 模板 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-thymeleaf</artifactId>
    </dependency>
    <!-- Lombok -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <!-- OkHttp（Stream Load 用）-->
    <dependency>
        <groupId>com.squareup.okhttp3</groupId>
        <artifactId>okhttp</artifactId>
        <version>4.12.0</version>
    </dependency>
</dependencies>

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:9030/demo?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    username: root
    password:
  thymeleaf:
    cache: false

doris:
  fe-host: http://127.0.0.1:8030
  stream-load-url: http://127.0.0.1:8030/api/demo/orders/_stream_load

-- 连接方式：mysql -uroot -P9030 -h127.0.0.1
CREATE DATABASE IF NOT EXISTS demo;
USE demo;

-- 功能1：订单明细表（主键模型，支持 UPSERT）
CREATE TABLE IF NOT EXISTS orders (
    order_id   BIGINT          NOT NULL COMMENT '订单ID',
    user_id    INT             NOT NULL COMMENT '用户ID',
    product_id INT             NOT NULL COMMENT '商品ID',
    amount     DECIMAL(10, 2)  NOT NULL COMMENT '金额',
    status     VARCHAR(20)     NOT NULL COMMENT '状态',
    created_at DATETIME        NOT NULL COMMENT '创建时间'
) ENGINE=OLAP
UNIQUE KEY(order_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 4
PROPERTIES ("replication_num" = "1");

-- 功能2：用户行为日志（明细模型 + 倒排索引）
CREATE TABLE IF NOT EXISTS user_events (
    event_time DATETIME     NOT NULL,
    user_id    INT          NOT NULL,
    event_type VARCHAR(50)  NOT NULL,
    page       VARCHAR(200) NOT NULL,
    extra      VARIANT                   -- 半结构化字段
) ENGINE=OLAP
DUPLICATE KEY(event_time, user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 4
PROPERTIES (
    "replication_num" = "1",
    "inverted_index_storage_format" = "V2"
);

-- 倒排索引（支持全文检索）
CREATE INDEX idx_page ON user_events(page) USING INVERTED;
CREATE INDEX idx_event_type ON user_events(event_type) USING INVERTED;

-- 功能3：销售汇总（聚合模型）
CREATE TABLE IF NOT EXISTS sales_summary (
    dt         DATE        NOT NULL,
    category   VARCHAR(50) NOT NULL,
    total_amt  DECIMAL(18,2) REPLACE,
    order_cnt  BIGINT      SUM
) ENGINE=OLAP
AGGREGATE KEY(dt, category)
DISTRIBUTED BY HASH(dt) BUCKETS 4
PROPERTIES ("replication_num" = "1");

-- 功能4：物化视图（加速报表查询）
CREATE MATERIALIZED VIEW mv_daily_sales
AS SELECT
    DATE(created_at) AS dt,
    status,
    COUNT(*) AS cnt,
    SUM(amount) AS total
FROM orders
GROUP BY DATE(created_at), status;

// OrderController.java
@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderController {

    private final JdbcTemplate jdbc;

    /** 查询订单列表（支持状态过滤） */
    @GetMapping
    public List<Map<String, Object>> listOrders(
            @RequestParam(defaultValue = "all") String status,
            @RequestParam(defaultValue = "20") int limit) {

        String sql = "all".equals(status)
            ? "SELECT * FROM demo.orders ORDER BY created_at DESC LIMIT ?"
            : "SELECT * FROM demo.orders WHERE status = ? ORDER BY created_at DESC LIMIT ?";

        return "all".equals(status)
            ? jdbc.queryForList(sql, limit)
            : jdbc.queryForList(sql, status, limit);
    }

    /** 写入/更新订单（UNIQUE KEY 自动做 UPSERT） */
    @PostMapping
    public Map<String, Object> upsertOrder(@RequestBody Map<String, Object> order) {
        String sql = "INSERT INTO demo.orders VALUES (?,?,?,?,?,NOW())";
        int rows = jdbc.update(sql,
            order.get("orderId"), order.get("userId"),
            order.get("productId"), order.get("amount"), order.get("status"));
        return Map.of("success", rows > 0, "msg", "UPSERT 成功");
    }

    /** 实时 GMV 聚合 */
    @GetMapping("/gmv")
    public Map<String, Object> gmv() {
        String sql = "SELECT COUNT(*) AS cnt, SUM(amount) AS gmv "
            + "FROM demo.orders WHERE created_at >= CURDATE()";
        return jdbc.queryForMap(sql);
    }
}

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org">
<head>
  <title>订单实时看板</title>
  <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css"/>
</head>
<body class="container mt-4">
  <h2>📦 订单实时看板</h2>
  <!-- GMV 卡片 -->
  <div class="row mb-4" id="gmvCards">
    <div class="col-md-3">
      <div class="card text-white bg-primary">
        <div class="card-body">
          <h5>今日 GMV</h5>
          <h2 id="gmvVal">加载中...</h2>
        </div>
      </div>
    </div>
  </div>
  <!-- 订单表格 -->
  <table class="table table-striped" id="orderTable">
    <thead><tr><th>订单ID</th><th>用户</th><th>金额</th><th>状态</th><th>时间</th></tr></thead>
    <tbody id="orderBody"></tbody>
  </table>
  <script>
    async function load() {
      const r = await fetch('/api/orders');
      const data = await r.json();
      document.getElementById('orderBody').innerHTML =
        data.map(o => `<tr><td>${o.order_id}</td><td>${o.user_id}</td>
          <td>${o.amount}</td><td>${o.status}</td><td>${o.created_at}</td></tr>`).join('');
      const g = await (await fetch('/api/orders/gmv')).json();
      document.getElementById('gmvVal').textContent = '¥' + Number(g.gmv||0).toFixed(2);
    }
    load(); setInterval(load, 5000);
  </script>
</body>
</html>

// EventController.java
@RestController
@RequestMapping("/api/events")
@RequiredArgsConstructor
public class EventController {

    private final JdbcTemplate jdbc;

    /** 全文检索（利用 Doris 倒排索引） */
    @GetMapping("/search")
    public List<Map<String, Object>> search(
            @RequestParam String keyword,
            @RequestParam(defaultValue = "50") int limit) {
        // MATCH_ANY 是 Doris 倒排索引全文检索语法
        String sql = "SELECT event_time, user_id, event_type, page "
            + "FROM demo.user_events "
            + "WHERE page MATCH_ANY ? "
            + "ORDER BY event_time DESC LIMIT ?";
        return jdbc.queryForList(sql, keyword, limit);
    }

    /** 批量写入事件（Stream Load 方式，高吞吐） */
    @PostMapping("/batch")
    public Map<String, Object> batchInsert(@RequestBody List<Map<String, Object>> events) {
        String sql = "INSERT INTO demo.user_events VALUES (?,?,?,?,?)";
        int[][] result = jdbc.batchUpdate(sql,
            events, events.size(),
            (ps, event) -> {
                ps.setString(1, event.get("eventTime").toString());
                ps.setInt(2, (Integer) event.get("userId"));
                ps.setString(3, event.get("eventType").toString());
                ps.setString(4, event.get("page").toString());
                ps.setString(5, event.getOrDefault("extra", "{}").toString());
            });
        return Map.of("inserted", events.size());
    }

    /** 事件类型分布统计 */
    @GetMapping("/stats")
    public List<Map<String, Object>> stats() {
        String sql = "SELECT event_type, COUNT(*) AS cnt "
            + "FROM demo.user_events "
            + "WHERE event_time >= DATE_SUB(NOW(), INTERVAL 1 HOUR) "
            + "GROUP BY event_type ORDER BY cnt DESC";
        return jdbc.queryForList(sql);
    }
}

// DorisStreamLoadService.java
@Service
public class DorisStreamLoadService {

    @Value("${doris.fe-host}")
    private String feHost;

    private final OkHttpClient client = new OkHttpClient();

    /**
     * Stream Load：高吞吐批量写入
     * 比 JDBC INSERT 快 5-10x，适合每批 1000+ 行
     */
    public String streamLoad(String db, String table, String csvData) throws Exception {
        String url = feHost + "/api/" + db + "/" + table + "/_stream_load";

        // Base64 认证（root 无密码）
        String credentials = Base64.getEncoder()
            .encodeToString("root:".getBytes(StandardCharsets.UTF_8));

        RequestBody body = RequestBody.create(
            csvData.getBytes(StandardCharsets.UTF_8),
            MediaType.parse("text/plain; charset=utf-8"));

        Request request = new Request.Builder()
            .url(url)
            .addHeader("Authorization", "Basic " + credentials)
            .addHeader("Expect", "100-continue")
            .addHeader("label", "stream-load-" + System.currentTimeMillis())
            .addHeader("format", "csv")
            .addHeader("column_separator", ",")
            .addHeader("columns", "order_id,user_id,product_id,amount,status,created_at")
            .put(body)
            .build();

        try (Response response = client.newCall(request).execute()) {
            return response.body().string();
        }
    }

    /** 生成测试数据并批量写入 */
    public void loadTestData(int count) throws Exception {
        StringBuilder sb = new StringBuilder();
        Random rand = new Random();
        String[] statuses = {"pending", "paid", "shipped", "done"};
        for (int i = 1; i <= count; i++) {
            sb.append(i).append(",")
              .append(rand.nextInt(1000) + 1).append(",")
              .append(rand.nextInt(100) + 1).append(",")
              .append(String.format("%.2f", rand.nextDouble() * 1000)).append(",")
              .append(statuses[rand.nextInt(statuses.length)]).append(",")
              .append("2025-06-12 10:00:00").append("\n");
        }
        String result = streamLoad("demo", "orders", sb.toString());
        System.out.println("Stream Load 结果: " + result);
    }
}

// ReportController.java
@RestController
@RequestMapping("/api/report")
@RequiredArgsConstructor
public class ReportController {

    private final JdbcTemplate jdbc;

    /**
     * 查询会自动命中物化视图 mv_daily_sales
     * 无需改 SQL，Doris 优化器透明改写
     */
    @GetMapping("/daily")
    public List<Map<String, Object>> dailyReport(
            @RequestParam(defaultValue = "7") int days) {
        String sql = "SELECT "
            + "  DATE(created_at) AS dt, "
            + "  status, "
            + "  COUNT(*) AS cnt, "
            + "  SUM(amount) AS total "
            + "FROM demo.orders "
            + "WHERE created_at >= DATE_SUB(NOW(), INTERVAL ? DAY) "
            + "GROUP BY DATE(created_at), status "
            + "ORDER BY dt DESC";
        return jdbc.queryForList(sql, days);
    }

    /** 验证是否命中物化视图 */
    @GetMapping("/explain")
    public String explainQuery() {
        String sql = "EXPLAIN SELECT DATE(created_at), status, COUNT(*), SUM(amount) "
            + "FROM demo.orders GROUP BY DATE(created_at), status";
        List<Map<String, Object>> rows = jdbc.queryForList(sql);
        return rows.stream()
            .map(r -> r.values().iterator().next().toString())
            .collect(Collectors.joining("\n"));
    }
}

-- 创建 MySQL 外部 Catalog（无需数据迁移直接查询）
CREATE CATALOG mysql_catalog PROPERTIES (
    "type" = "jdbc",
    "user" = "root",
    "password" = "yourpwd",
    "jdbc_url" = "jdbc:mysql://mysql-host:3306",
    "driver_url" = "https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.33/mysql-connector-java-8.0.33.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver"
);

-- 跨库联邦查询（Doris 本地 + MySQL 外表 JOIN）
SELECT o.order_id, o.amount, u.username, u.email
FROM demo.orders o
JOIN mysql_catalog.userdb.users u ON o.user_id = u.id
WHERE o.created_at >= CURDATE();

# 假设三台机器：node1(10.0.0.1) node2(10.0.0.2) node3(10.0.0.3)

# ── Step 1：所有节点准备 ──
sudo sysctl -w vm.max_map_count=2000000
sudo sysctl -w vm.swappiness=0
echo 'vm.max_map_count=2000000' | sudo tee -a /etc/sysctl.conf

# ── Step 2：下载并解压（以 node1 为例）──
wget https://apache-doris-releases.oss-accelerate.aliyuncs.com/apache-doris-3.0.6-bin-x64.tar.gz
tar -zxvf apache-doris-3.0.6-bin-x64.tar.gz

# ── Step 3：配置 FE（fe/conf/fe.conf）──
echo 'meta_dir = /data/doris/fe-meta' >> fe/conf/fe.conf
echo 'priority_networks = 10.0.0.0/24' >> fe/conf/fe.conf

# ── Step 4：启动第一个 FE（Leader）──
bash fe/bin/start_fe.sh --daemon

# ── Step 5：注册其他 FE 为 Follower ──
mysql -uroot -P9030 -h10.0.0.1 \
  -e "ALTER SYSTEM ADD FOLLOWER '10.0.0.2:9010'"
mysql -uroot -P9030 -h10.0.0.1 \
  -e "ALTER SYSTEM ADD FOLLOWER '10.0.0.3:9010'"

# ── Step 6：在 node2/node3 启动 FE，指定 helper ──
bash fe/bin/start_fe.sh --helper 10.0.0.1:9010 --daemon

# ── Step 7：配置 BE（be/conf/be.conf）──
echo 'storage_root_path = /data/doris/be-storage' >> be/conf/be.conf
echo 'priority_networks = 10.0.0.0/24' >> be/conf/be.conf

# ── Step 8：注册并启动三个 BE ──
for host in 10.0.0.1 10.0.0.2 10.0.0.3; do
  mysql -uroot -P9030 -h10.0.0.1 \
    -e "ALTER SYSTEM ADD BACKEND '${host}:9050'"
done
# 在每台机器上执行
bash be/bin/start_be.sh --daemon

# ── Step 9：验证集群状态 ──
mysql -uroot -P9030 -h10.0.0.1 \
  -e "SHOW PROC '/frontends'"
mysql -uroot -P9030 -h10.0.0.1 \
  -e "SHOW PROC '/backends'"

# Ubuntu 22.04
curl https://clickhouse.com/ | sh
./clickhouse install
sudo clickhouse start
clickhouse-client  # 进入 CLI

docker run -d --name starrocks \
  -p 9030:9030 -p 8030:8030 -p 8040:8040 \
  starrocks/allin1-ubuntu:3.3-latest
# 等待约 30 秒后
mysql -P9030 -h127.0.0.1 -uroot --prompt="StarRocks> "

-- 存储 AI 代码审查日志
CREATE TABLE code_review_log (
    review_id   BIGINT      NOT NULL,
    repo        VARCHAR(200),
    file_path   VARCHAR(500),
    language    VARCHAR(50),
    issue_type  VARCHAR(100),
    ai_model    VARCHAR(50),
    score       FLOAT,
    suggestion  TEXT,        -- AI 生成的建议
    reviewed_at DATETIME
) ENGINE=OLAP
UNIQUE KEY(review_id)
DISTRIBUTED BY HASH(review_id) BUCKETS 8
PROPERTIES ("replication_num" = "1");

-- 实时查询：哪些文件 Bug 最多？
SELECT file_path, COUNT(*) AS issue_cnt, AVG(score) AS avg_score
FROM code_review_log
WHERE reviewed_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
GROUP BY file_path
ORDER BY issue_cnt DESC
LIMIT 20;

# 伪代码：LangChain + Doris 混合检索
def hybrid_search(user_question: str) -> str:
    # 1. 向量数据库（Chroma/Milvus）做语义检索
    semantic_docs = vector_store.similarity_search(user_question, k=5)
    
    # 2. Doris 做结构化精确过滤（补充向量检索盲区）
    sql = """
        SELECT content, source, updated_at
        FROM knowledge_base
        WHERE content MATCH_ANY %s
          AND category = 'technical'
        ORDER BY updated_at DESC
        LIMIT 10
    """
    doris_docs = doris_client.execute(sql, [user_question])
    
    # 3. 合并上下文送给 LLM
    context = semantic_docs + doris_docs
    return llm.chat(f"基于以下资料回答问题: {context}\n问题: {user_question}")