从零到一：Flink与Doris的完美邂逅——数据流导入实战指南-开发者社区

从零到一：Flink与Doris的完美邂逅——数据流导入实战指南

1. 实时数据处理的黄金组合

在当今数据驱动的商业环境中，电商平台需要实时处理海量订单数据以支持即时决策。Apache Flink作为流处理引擎的佼佼者，与Apache Doris这一高性能MPP分析型数据库的结合，为实时数据分析提供了完美的技术栈。

为什么选择Flink+Doris？

Flink提供**精确一次（Exactly-Once）**的流处理语义
Doris实现亚秒级的查询响应
两者结合可实现从数据摄入到分析的端到端实时管道

我曾在一个跨境电商项目中采用这套方案，将订单分析延迟从小时级降低到秒级，促销活动的库存预警响应速度提升了20倍。

2. 环境准备与依赖配置

2.1 基础环境要求

确保已安装：

JDK 8/11
Flink 1.16+集群
Doris 1.0+集群
Maven 3.6+

2.2 Maven依赖配置

在pom.xml中添加最新connector依赖：

<dependency> <groupId>org.apache.doris</groupId> <artifactId>flink-doris-connector-1.16</artifactId> <version>1.6.0</version> </dependency>

版本兼容性参考：

Connector版本	Flink版本	Doris版本
1.4.x	1.15-1.17	≥1.0
1.5.x	1.16	≥1.0
1.6.x	1.16	≥1.0

3. Doris表设计与准备

3.1 创建订单分析表

CREATE TABLE IF NOT EXISTS order_analysis.realtime_orders ( `order_id` VARCHAR(64) NOT NULL COMMENT "订单ID", `user_id` LARGEINT NOT NULL COMMENT "用户ID", `product_id` BIGINT COMMENT "商品ID", `order_time` DATETIME COMMENT "下单时间", `payment_amount` DECIMAL(12,2) SUM DEFAULT "0" COMMENT "支付金额", `payment_method` TINYINT COMMENT "支付方式", `province_code` INT COMMENT "省份编码" ) ENGINE=OLAP AGGREGATE KEY(`order_id`, `user_id`, `product_id`, `order_time`) PARTITION BY RANGE(`order_time`)( PARTITION p202301 VALUES LESS THAN ('2023-02-01'), PARTITION p202302 VALUES LESS THAN ('2023-03-01') ) DISTRIBUTED BY HASH(`order_id`) BUCKETS 8 PROPERTIES ( "replication_allocation" = "tag.location.default: 3", "dynamic_partition.enable" = "true", "dynamic_partition.time_unit" = "MONTH", "dynamic_partition.start" = "-12", "dynamic_partition.end" = "3" );

关键设计要点：

使用AGGREGATE KEY模型适合指标汇总场景
动态分区自动管理时间分区
Bucket数量建议为BE节点数的3-5倍

4. Flink数据流开发实战

4.1 基础数据流写入

public class OrderStreamJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000); // 10秒checkpoint // 模拟订单数据源 DataStreamSource<String> orderStream = env.addSource(new OrderMockSource()); // 构建Doris Sink DorisSink<String> dorisSink = DorisSink.<String>builder() .setDorisOptions(DorisOptions.builder() .setFenodes("doris-fe:8030") .setTableIdentifier("order_analysis.realtime_orders") .setUsername("flink_user") .setPassword("flink@123") .build()) .setDorisExecutionOptions(DorisExecutionOptions.builder() .setLabelPrefix("order-sync-") .setDeletable(false) .setStreamLoadProp(getStreamLoadProps()) .build()) .setSerializer(new SimpleStringSerializer()) .build(); // 数据写入 orderStream.sinkTo(dorisSink); env.execute("Order Stream to Doris"); } private static Properties getStreamLoadProps() { Properties props = new Properties(); props.setProperty("column_separator", "\t"); props.setProperty("columns", "order_id,user_id,product_id," + "order_time,payment_amount,payment_method,province_code"); return props; } }

4.2 高级特性应用

JSON格式数据写入：

Properties jsonProps = new Properties(); jsonProps.setProperty("format", "json"); jsonProps.setProperty("read_json_by_line", "true"); DorisExecutionOptions execOptions = DorisExecutionOptions.builder() .setLabelPrefix("json-orders-") .setStreamLoadProp(jsonProps) .build(); // RowData序列化配置 String[] fields = {"order_id","user_id","product_id","order_time", "payment_amount","payment_method","province_code"}; DataType[] types = {DataTypes.STRING(), DataTypes.BIGINT(), DataTypes.BIGINT(), DataTypes.TIMESTAMP(), DataTypes.DECIMAL(12,2), DataTypes.TINYINT(), DataTypes.INT()}; RowDataSerializer serializer = RowDataSerializer.builder() .setFieldNames(fields) .setType("json") .setFieldType(types) .build();

5. 生产环境最佳实践

5.1 性能调优指南

关键参数配置：

参数	建议值	说明
sink.batch.size	5000-10000	批次大小
sink.batch.interval	10s	批次间隔
checkpoint.interval	30s	Checkpoint间隔
parallelism	BE节点数*2	并行度设置

常见问题处理：

注意：遇到"Label has already been used"错误时，需要确保：
从checkpoint恢复时不要修改labelPrefix
非正常停止后需等待事务超时(默认1小时)或修改FE配置

5.2 监控与运维

关键监控指标：

Flink: checkpoint持续时间/失败次数

Doris:

SHOW PROC '/stream_load'; SHOW ROUTINE LOAD WHERE NAME = 'your_job';

运维建议：

为Flink作业单独配置Doris用户和资源隔离
定期清理已完成的事务记录
监控BE节点的内存和IO使用率

6. 典型应用场景扩展

6.1 实时订单看板

-- Doris物化视图加速查询 CREATE MATERIALIZED VIEW order_dashboard_mv DISTRIBUTED BY HASH(province_code) REFRESH ASYNC AS SELECT province_code, DATE_FORMAT(order_time, '%Y-%m-%d %H:00') AS hour_time, COUNT(DISTINCT user_id) AS uv, SUM(payment_amount) AS gmv FROM order_analysis.realtime_orders GROUP BY province_code, hour_time;

6.2 实时风控系统

// 使用Flink CEP检测异常订单 Pattern<OrderEvent, ?> riskPattern = Pattern.<OrderEvent>begin("start") .where(new SimpleCondition<OrderEvent>() { @Override public boolean filter(OrderEvent value) { return value.getAmount() > 10000; } }) .next("follow") .within(Time.minutes(5)); CEP.pattern(orderStream.keyBy(OrderEvent::getUserId), riskPattern) .process(new RiskAlertProcessFunction()) .addSink(new DorisAlertSink());

在实际项目中，这套方案帮助我们识别了超过80%的欺诈订单，平均延迟控制在3秒以内。