亚秒级响应！Apache Doris亿级数据实时分析实战拆解

-- 实战案例：电商订单表设计
CREATE TABLE order_analysis (
  order_id BIGINT,
  category_id INT,
  price DECIMAL(10,2),
  order_time DATETIME
) ENGINE=olap
PARTITION BY RANGE(order_time) (  -- 时间分区加速冷热数据分离
  PARTITION p202301 VALUES [('2023-01-01'), ('2023-02-01')),
  PARTITION p202302 VALUES [('2023-02-01'), ('2023-03-01'))
)
DISTRIBUTED BY HASH(category_id) BUCKETS 10  -- 分桶避免数据倾斜
PROPERTIES ("replication_num" = "3");

CREATE TABLE user_behavior (
    user_id BIGINT,
    last_login DATETIME,
    UNIQUE KEY(user_id)
) ENGINE=olap

-- 建表：类目交易流水表
CREATE TABLE category_sales (
  ts DATETIME,         -- 事件时间
  category_id INT,     -- 类目ID
  sku_id BIGINT,       -- 商品ID
  amount DECIMAL(12,2),-- 交易金额
  province VARCHAR(20) -- 省份
) ENGINE=olap
DUPLICATE KEY(ts, category_id)  -- 允许重复记录
PARTITION BY RANGE(ts) (        -- 按小时分区
  PARTITION p202306 VALUES [('2023-06-01'), ('2023-07-01'))
)
DISTRIBUTED BY HASH(category_id) BUCKETS 12  
PROPERTIES (
  "storage_policy" = "SSD",      -- 热数据存SSD
  "replication_num" = "2" 
);

-- 创建预聚合视图（自动命中查询）
ALTER TABLE category_sales ADD ROLLUP sales_summary (
  ts, category_id, province, 
  SUM(amount)      -- 预计算类目销售额
);

/* 获取各省TOP3热卖类目 */
SELECT province, category_id, SUM(amount) AS total
FROM category_sales 
WHERE ts > NOW() - INTERVAL 1 HOUR  -- 实时过滤
GROUP BY province, category_id
ORDER BY total DESC 
LIMIT 3; 

-- 用户行为日志表（Kafka实时接入）
CREATE TABLE user_behavior (
  log_time DATETIME,
  user_id BIGINT,
  item_id BIGINT,
  action_type VARCHAR(10) -- click/purchase
) ENGINE=olap
UNIQUE KEY(log_time, user_id, item_id)  -- 主键更新模型
PARTITION BY RANGE(log_time) (
  PARTITION p202305 VALUES [('2023-05-01'), ('2023-06-01'))
)
DISTRIBUTED BY HASH(item_id) BUCKETS 8;

-- 商品热度物化视图（分钟级更新）
CREATE MATERIALIZED VIEW item_hotness AS 
SELECT 
  item_id, 
  COUNT_IF(action_type='click') AS clicks, -- 点击量
  COUNT_IF(action_type='purchase') AS buys -- 购买量
FROM user_behavior 
GROUP BY item_id;

/* 计算动态调价系数 */
SELECT 
  item_id,
  clicks / (buys + 1) AS click_ratio, -- 防除零
  CASE 
    WHEN clicks > 1000 AND click_ratio < 0.1 THEN 1.1  -- 高热度低转化：涨价10%
    WHEN clicks < 100 AND buys > 50 THEN 0.9            -- 低热度高转化：降价10%
    ELSE 1.0
  END AS price_factor
FROM item_hotness 
WHERE log_time > NOW() - INTERVAL 5 MINUTE; -- 滑动窗口

-- 用户画像表优化案例
CREATE TABLE user_profiles (
    user_id BIGINT,
    age TINYINT,
    gender CHAR(1),
    city_code SMALLINT,
    last_active DATETIME,
    INDEX idx_city_gender (city_code, gender)  -- 复合前缀索引
) ENGINE=olap
DUPLICATE KEY(user_id)

ALTER TABLE order_detail SET (
  "storage_policy" = "HOT:3 DAYS COLD:30 DAYS",  -- 3天热数据存SSD
  "storage_cooldown_time" = "2023-12-31 00:00:00" -- 冷数据自动转HDD
);

// Java示例：Stream Load批量导入（百万级/秒）
HttpClient client = HttpClient.newHttpClient();
String loadUrl = "http://fe_host:8030/api/db/table/_stream_load";
String data = "1\tproductA\t100.0\n2\tproductB\t200.0"; // TSV格式

HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create(loadUrl))
    .header("format", "csv")
    .header("column_separator", "\t")
    .header("columns", "id,name,price")  // 列映射
    .POST(HttpRequest.BodyPublishers.ofString(data))
    .build();

HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
System.out.println(response.body()); // 查看导入结果

CREATE ROUTINE LOAD job_order_kafka ON order_stream
PROPERTIES (
  "desired_concurrent_number"="8",     -- 并发消费线程
  "max_batch_interval" = "20",         -- 最大攒批时间(秒)
  "max_batch_rows" = "200000",         -- 单批次最大行数
  "strict_mode" = "false"              -- 容忍类型转换错误
) 
FROM KAFKA (
  "kafka_broker_list" = "broker1:9092",
  "kafka_topic" = "order_topic",
  "property.group.id" = "doris_consumer"
);

-- 创建资源组限制大查询
CREATE RESOURCE GROUP report_group
PROPERTIES (
  "cpu_share"="10",       -- CPU权重
  "memory_limit"="30%",    -- 内存上限
  "max_concurrency"="5"   -- 并发限制
);

-- 将BI用户绑定到资源组
SET PROPERTY FOR 'bi_user' 'resource_group' = 'report_group';

# 未来方向示例：AI驱动查询优化器
from doris.ai import QueryAdvisor

advisor = QueryAdvisor(model="gpt-4o")
sql = "SELECT user_id, SUM(price) FROM orders WHERE dt>='2024-01-01' GROUP BY user_id"

# 智能建议生成
advice = advisor.optimize(sql)
print(advice.feedback)
# 输出："建议创建ROLLUP( user_id, price )，预计查询速度提升15x"