Layered Real‑Time Data Warehouse with Flink CDC, Kafka & Doris

Problem Overview

Many teams build a simple pipeline MySQL → Flink CDC → Kafka → Flink → Doris. It works in a demo environment, but in production it quickly runs into issues such as CDC checkpoint timeouts, Kafka topic lag, dimension‑table joins that overload the source database, frequent small‑batch imports to Doris, and unclear exactly‑once semantics.

Why a Layered Design

Real‑time warehouses still need the classic ODS‑DWD‑DWS‑ADS layers, not to store duplicate copies but to separate responsibilities:

ODS : raw change events, primary key, operation type, event time; serves as a replayable buffer.

DWD : transforms raw events into business‑level facts, performs deduplication, state normalization and dimension enrichment.

DWS : common aggregations (minute, hour, shop, channel) that avoid repeated calculations downstream.

ADS : final tables tailored for dashboards, reports or APIs, optimized for query performance.

This separation makes the pipeline easier to evolve, scale and recover.

Technology Choices

Flink CDC reads MySQL binlog, supports an initial snapshot and seamless incremental capture, and integrates with Flink’s checkpointing. Kafka provides buffering, decoupling, replay capability and ordered partitions. Doris offers low‑latency OLAP queries, supports high‑concurrency writes and is MySQL‑compatible, making it a practical sink for real‑time analytics.

Key Implementation Details

ODS pipeline (SQL DDL) :

CREATE TABLE mysql_orders (
    order_id BIGINT,
    user_id BIGINT,
    shop_id BIGINT,
    product_id BIGINT,
    order_status STRING,
    total_amount DECIMAL(16,2),
    source_channel STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql-primary',
    'port' = '3306',
    'username' = 'cdc_user',
    'password' = '${CDC_PASSWORD}',
    'database-name' = 'trade',
    'table-name' = 'orders',
    'scan.startup.mode' = 'initial',
    'server-time-zone' = 'Asia/Shanghai'
);

CREATE TABLE kafka_ods_orders (
    `before` ROW<order_id BIGINT, user_id BIGINT, shop_id BIGINT, product_id BIGINT, order_status STRING, total_amount DECIMAL(16,2), source_channel STRING, created_at STRING, updated_at STRING>,
    `after`  ROW<order_id BIGINT, user_id BIGINT, shop_id BIGINT, product_id BIGINT, order_status STRING, total_amount DECIMAL(16,2), source_channel STRING, created_at STRING, updated_at STRING>,
    op STRING,
    ts_ms BIGINT
) WITH (
    'connector' = 'kafka',
    'topic' = 'ods.mysql.trade.orders',
    'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    'format' = 'debezium-json'
);

INSERT INTO kafka_ods_orders SELECT * FROM mysql_orders;

DWD job (Java) – job split into three responsibilities:

realtime-warehouse/
├── job-cdc-ods/
│   └── CdcToOdsJob.java
├── job-dwd-order/
│   ├── DwdOrderJob.java
│   ├── function/OrderStateNormalizeProcessFunction.java
│   └── dim/UserDimAsyncFunction.java
├── job-dws-trade/
│   └── DwsTradeMetricJob.java
└── deploy/...

Example of the CDC‑to‑ODS job configuration:

public static void main(String[] args) throws Exception {
    JobConfig config = JobConfigLoader.load(args);
    StreamExecutionEnvironment env = FlinkEnvFactory.createBaseEnv(config.getCheckpointIntervalSeconds());

    KafkaSource<OrderFactEvent> orderSource = KafkaSource.<OrderFactEvent>builder()
        .setBootstrapServers(config.getKafkaBootstrapServers())
        .setTopics(config.getOdsOrdersTopic())
        .setGroupId("dwd-order-job")
        .setStartingOffsets(OffsetsInitializer.committedOffsets())
        .setValueOnlyDeserializer(new OrderFactEventDeserializationSchema())
        .build();

    DataStream<OrderFactEvent> source = env.fromSource(orderSource,
        WatermarkStrategy.<OrderFactEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((e, ts) -> e.getEventTime()), "ods-orders-source");

    SingleOutputStreamOperator<OrderFactEvent> cleaned = source
        .filter(DwdValidators::isValid)
        .keyBy(OrderFactEvent::getOrderId)
        .process(new OrderStateNormalizeProcessFunction())
        .name("normalize-order-state");

    DataStream<OrderFactEvent> enriched = AsyncDataStream.unorderedWait(
        cleaned,
        new UserDimAsyncFunction(),
        config.getAsyncDimTimeoutSeconds(),
        TimeUnit.SECONDS,
        config.getAsyncDimCapacity())
        .name("async-user-dim");

    KafkaSink<OrderFactEvent> sink = KafkaSink.<OrderFactEvent>builder()
        .setBootstrapServers(config.getKafkaBootstrapServers())
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
            .setTopic(config.getDwdOrderFactTopic())
            .setValueSerializationSchema(new JsonSerializationSchema<>())
            .build())
        .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();

    enriched.sinkTo(sink).name("dwd-order-fact-sink");
    env.execute("dwd-order-job");
}

Async dimension enrichment (user dimension) shows how to handle timeouts and thread‑pool limits:

public class UserDimAsyncFunction extends RichAsyncFunction<OrderFactEvent, OrderFactEvent> {
    private transient ExecutorService executorService;
    private transient UserDimRepository userDimRepository;

    @Override
    public void open(Configuration parameters) {
        executorService = Executors.newFixedThreadPool(16);
        userDimRepository = new UserDimRepository();
    }

    @Override
    public void asyncInvoke(OrderFactEvent input, ResultFuture<OrderFactEvent> resultFuture) {
        CompletableFuture.supplyAsync(() -> userDimRepository.queryByUserId(input.getUserId()), executorService)
            .whenComplete((userDim, ex) -> {
                if (ex != null) {
                    resultFuture.complete(Collections.singleton(input));
                    return;
                }
                if (userDim != null) {
                    input.setUserLevel(userDim.userLevel());
                    input.setCityCode(userDim.cityCode());
                } else {
                    input.setUserLevel("UNKNOWN");
                    input.setCityCode("UNKNOWN");
                }
                resultFuture.complete(Collections.singleton(input));
            });
    }

    @Override
    public void timeout(OrderFactEvent input, ResultFuture<OrderFactEvent> resultFuture) {
        input.setUserLevel("TIMEOUT");
        input.setCityCode("TIMEOUT");
        resultFuture.complete(Collections.singleton(input));
    }

    @Override
    public void close() throws Exception {
        if (executorService != null) {
            executorService.shutdown();
        }
    }
}

State normalization (order status) ensures idempotent downstream facts:

public class OrderStateNormalizeProcessFunction extends KeyedProcessFunction<Long, OrderFactEvent, OrderFactEvent> {
    private transient ValueState<String> latestStatusState;

    @Override
    public void open(Configuration parameters) {
        latestStatusState = getRuntimeContext().getState(new ValueStateDescriptor<>("latest-status", String.class));
    }

    @Override
    public void processElement(OrderFactEvent value, Context ctx, Collector<OrderFactEvent> out) throws Exception {
        String previous = latestStatusState.value();
        String current = value.getOrderStatus();
        if (previous != null && previous.equals(current) && "u".equals(value.getOpType())) {
            return; // duplicate update, drop
        }
        latestStatusState.update(current);
        out.collect(value);
    }
}

Flink SQL aggregation (DWS → ADS) demonstrates windowed metrics for shop‑minute level:

CREATE TABLE dwd_order_fact (
    order_id BIGINT,
    shop_id BIGINT,
    source_channel STRING,
    pay_status STRING,
    pay_amount DECIMAL(16,2),
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'dwd.trade.order_fact',
    ...
);

CREATE TABLE ads_shop_trade_minute (
    metric_time TIMESTAMP,
    shop_id BIGINT,
    source_channel STRING,
    paid_order_count BIGINT,
    paid_amount DECIMAL(18,2),
    updated_at TIMESTAMP
) WITH (
    'connector' = 'doris',
    'fenodes' = 'doris-fe-1:8030,doris-fe-2:8030',
    'table.identifier' = 'rt_ads.ads_shop_trade_minute',
    ...
);

INSERT INTO ads_shop_trade_minute
SELECT
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS metric_time,
    shop_id,
    source_channel,
    COUNT(DISTINCT CASE WHEN pay_status = 'PAID' THEN order_id END) AS paid_order_count,
    SUM(CASE WHEN pay_status = 'PAID' THEN pay_amount ELSE 0 END) AS paid_amount,
    CURRENT_TIMESTAMP AS updated_at
FROM dwd_order_fact
GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE), shop_id, source_channel;

Doris table DDL for the minute‑level metric table:

CREATE TABLE ads_shop_trade_minute (
    metric_time DATETIME NOT NULL,
    shop_id BIGINT NOT NULL,
    source_channel VARCHAR(32) NOT NULL,
    paid_order_count BIGINT SUM DEFAULT "0",
    paid_amount DECIMAL(18,2) SUM DEFAULT "0",
    updated_at DATETIME NOT NULL
) AGGREGATE KEY(metric_time, shop_id, source_channel)
PARTITION BY RANGE(metric_time)()
DISTRIBUTED BY HASH(shop_id) BUCKETS 16
PROPERTIES (
    "replication_allocation" = "tag.location.default: 3"
);

Operational Practices cover fault‑tolerance, back‑pressure handling, hot‑key mitigation, state size control (TTL for join and deduplication state), async I/O timeout handling, checkpoint configuration, and monitoring key metrics such as CDC delay, Kafka lag, checkpoint success rate, async I/O timeout rate, Doris import latency, dirty‑data count, and late‑data ratio.

Deployment on Kubernetes uses the FlinkOperator CRD; a minimal YAML example is provided to illustrate resource sizing, checkpoint settings, and upgrade mode (savepoint).

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: dws-trade-metric-job
spec:
  image: registry.example.com/realtime-warehouse:1.0.0
  flinkVersion: v1_18
  serviceAccount: flink
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    state.backend: rocksdb
    execution.checkpointing.interval: "60s"
    execution.checkpointing.timeout: "600s"
    execution.checkpointing.max-concurrent-checkpoints: "1"
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
  job:
    jarURI: local:///opt/flink/usrlib/job-dws-trade.jar
    entryClass: com.example.rtw.job.dws.DwsTradeMetricJob
    parallelism: 8
    upgradeMode: savepoint
    state: running

Overall, the article stresses that the success of a real‑time data warehouse depends less on the choice of components and more on clear layer responsibilities, idempotent design, replay capability, resource isolation, and comprehensive observability.