Flink + Doris + Airflow 构建企业级实时报表统计平台方案

目录

1. 背景与目标

2. 整体架构

3. 核心组件说明

3.1 Apache Flink

3.2 Apache Doris（原 Palo）

3.3 Apache Airflow

4. 实施步骤

步骤一：环境部署

步骤二：数据接入层设计

实时链路（Flink CDC + Kafka）

离线链路（Airflow 调度 Flink Batch Job）

步骤三：Flink 实时计算层

场景：实时销售大屏（每分钟聚合）

步骤四：Doris 表模型设计

表1：实时明细表（Duplicate Key 模型）

表2：分钟级聚合表（Aggregate Key 模型）

步骤五：Airflow 统一调度

DAG 示例：daily_report_dag.py

步骤六：报表与监控

报表层

监控体系

5. 方案优势对比（vs DataX 方案）

6. 最佳实践与注意事项

6.1 性能优化

6.2 数据一致性

6.3 成本控制

6.4 安全合规

7. 总结

1. 背景与目标

随着业务对数据时效性要求的提升，传统 T+1 的离线报表已无法满足实时决策需求。本方案基于 Apache Flink（流批一体计算引擎） + Apache Doris（高性能 MPP 分析型数据库） + Apache Airflow（工作流调度平台），构建一套支持 实时 + 离线融合 的企业级报表统计体系，实现：

• 实时数据接入与处理（秒级延迟）
• 高并发、低延迟的即席查询能力
• 统一调度管理批处理与流处理任务
• 支撑多维分析、大屏监控、运营报表等场景

2. 整体架构

[业务系统] [日志采集] [消息队列]
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌──────────┐ ┌────────────┐
│ MySQL Binlog│ │ FileBeat │ │ Kafka/Pulsar│
└──────┬──────┘ └─────┬────┘ └──────┬─────┘
│ │ │
└───────────────┴──────────────┘
│
┌─────────▼──────────┐
│ Apache Flink │ ←── Airflow 触发批作业
│ (实时计算 + 批处理) │
└─────────┬──────────┘
│
┌───────────▼────────────┐
│ Apache Doris │
│ (统一存储与分析引擎) │
└───────────┬────────────┘
│
┌────────────▼────────────┐
│ BI 工具 / 报表系统 / API│
│ (Superset, FineReport…)│
└─────────────────────────┘

3. 核心组件说明

3.1 Apache Flink

• 流批一体：同一套 API 处理实时流（Kafka）和历史批数据（HDFS/MySQL）
• 状态管理 & 容错：Exactly-once 语义保障数据一致性
• 丰富连接器：支持 Kafka、MySQL CDC、Doris、JDBC、File 等
• 窗口计算：支持滚动、滑动、会话窗口，适用于实时聚合

3.2 Apache Doris（原 Palo）

• MPP 架构：向量化执行引擎，高并发低延迟查询
• 实时写入：支持 Stream Load、Routine Load（Kafka 自动消费）、Broker Load
• 统一模型：明细模型 + 聚合模型 + 更新模型，适配不同场景
• 兼容 MySQL 协议：BI 工具可直接连接，无缝集成

3.3 Apache Airflow

• DAG 编排：可视化定义任务依赖关系
• 调度精准：基于 cron 表达式，支持 backfill
• 扩展性强：自定义 Operator（如 FlinkSubmitOperator、DorisQueryOperator）
• 监控告警：集成 Prometheus + Grafana 或邮件/钉钉通知

4. 实施步骤

步骤一：环境部署

建议使用 Kubernetes + Helm 统一编排，便于扩缩容与运维。

步骤二：数据接入层设计

实时链路（Flink CDC + Kafka）

// 示例：MySQL → Kafka（通过 Flink CDC）
CREATE TABLE mysql_orders (
id BIGINT,
order_no STRING,
amount DECIMAL(10,2),
create_time TIMESTAMP(3),
proc_time AS PROCTIME()
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'mysql-host',
'port' = '3306',
'username' = 'user',
'password' = 'pwd',
'database-name' = 'order_db',
'table-name' = 'orders'
);

— 写入 Kafka
INSERT INTO kafka_orders_topic
SELECT id, order_no, amount, create_time FROM mysql_orders;

离线链路（Airflow 调度 Flink Batch Job）

• 每日凌晨同步历史全量数据（如用户维度表）
• 使用 FlinkRunOperator 提交批作业 JAR 包

步骤三：Flink 实时计算层

场景：实时销售大屏（每分钟聚合）

— Flink SQL 作业
CREATE TABLE doris_sales_realtime (
stat_minute STRING,
city STRING,
order_cnt BIGINT,
total_amount DECIMAL(18,2)
) WITH (
'connector' = 'doris',
'fenodes' = 'fe:8030',
'table.identifier' = 'report.sales_realtime',
'username' = 'root',
'password' = '',
'sink.label-prefix' = 'sales_'
);

INSERT INTO doris_sales_realtime
SELECT
DATE_FORMAT(TUMBLE_ROWTIME(create_time, INTERVAL '1' MINUTE), 'yyyy-MM-dd HH:mm') AS stat_minute,
user_city AS city,
COUNT(*) AS order_cnt,
SUM(amount) AS total_amount
FROM kafka_orders_with_user_info
GROUP BY TUMBLE(create_time, INTERVAL '1' MINUTE), user_city;

注：通过维表 JOIN 补充用户城市信息（使用 Temporal Join）

步骤四：Doris 表模型设计

表1：实时明细表（Duplicate Key 模型）

CREATE TABLE orders_detail (
order_id BIGINT,
order_no VARCHAR(64),
user_id BIGINT,
amount DECIMAL(10,2),
create_time DATETIME
) ENGINE=OLAP
DUPLICATE KEY(order_id)
PARTITION BY RANGE(create_time) (
START ("2025-01-01") END ("2026-01-01") EVERY (INTERVAL 1 DAY)
)
DISTRIBUTED BY HASH(order_id) BUCKETS 10;

表2：分钟级聚合表（Aggregate Key 模型）

CREATE TABLE sales_realtime (
stat_minute DATETIME,
city VARCHAR(32),
order_cnt BIGINT SUM,
total_amount DECIMAL(18,2) SUM
) ENGINE=OLAP
AGGREGATE KEY(stat_minute, city)
PARTITION BY RANGE(stat_minute) (…)
DISTRIBUTED BY HASH(city) BUCKETS 8;

步骤五：Airflow 统一调度

DAG 示例：daily_report_dag.py

from airflow import DAG
from airflow.providers.apache.flink.operators.flink import FlinkRunOperator
from datetime import datetime, timedelta

default_args = {
'owner': 'data-team',
'retries': 2,
'retry_delay': timedelta(minutes=5)
}

with DAG(
'daily_sales_summary',
default_args=default_args,
schedule_interval='0 2 * * *', # 每天凌晨2点
start_date=datetime(2025, 11, 1),
catchup=False
) as dag:

# 1. 同步用户维度表（批处理）
sync_dim_user = FlinkRunOperator(
task_id='sync_dim_user',
application_args=['–mode=batch', '–table=dim_user'],
flink_cli_conn_id='flink_cluster'
)

# 2. 触发 Doris 物化视图刷新（可选）
refresh_mv = BashOperator(
task_id='refresh_doris_mv',
bash_command='curl -u root: -X POST "http://fe:8030/api/report/sales_daily/_load" …'
)

sync_dim_user >> refresh_mv

可集成 Flink Kubernetes Operator 实现更云原生的作业提交。

步骤六：报表与监控

报表层

• BI 工具：Superset / FineReport / Tableau 直连 Doris
• API 服务：Spring Boot + MyBatis 查询 Doris，供前端调用
• 大屏展示：ECharts + WebSocket 实时推送（从 Doris 轮询或 Kafka 推送）

监控体系

5. 方案优势对比（vs DataX 方案）

6. 最佳实践与注意事项

6.1 性能优化

• Flink：合理设置 state TTL、checkpoint 间隔、并行度
• Doris：分区裁剪、物化视图、Colocate Join
• Kafka：合理设置分区数，避免数据倾斜

6.2 数据一致性

• 实时链路：Flink + Doris 支持 Exactly-once（需开启两阶段提交）
• 离线链路：Airflow 任务幂等设计，避免重复写入

6.3 成本控制

• 冷热分离：Doris 支持冷数据自动迁移到对象存储（2.0+）
• 资源隔离：Flink on K8s 按业务划分命名空间

6.4 安全合规

• Doris 开启 LDAP 认证
• Kafka 启用 SASL/SSL
• Airflow RBAC 权限控制

7. 总结

Flink + Doris + Airflow 是面向未来的现代化数据架构组合：

• Flink 作为统一计算引擎，打通实时与离线；
• Doris 作为统一存储与查询引擎，简化 Lambda 架构；
• Airflow 作为调度中枢，实现端到端任务治理。

该方案适用于对 数据时效性、查询性能、系统稳定性 有较高要求的中大型企业，可支撑从运营报表到实时风控、用户画像等复杂场景。

适用阶段：企业数据中台建设、实时数仓升级、BI 系统重构