Flink + ClickHouse 实时数仓分层架构设计与核心实现（开源技术栈）

业务日志 → Nginx → Filebeat → Kafka (ODS)
                                    ↓
                            Flink SQL (ETL + 维表 JOIN)
                                    ↓
                           Kafka (DWD 明细流)
                                    ↓
                    Flink SQL (窗口聚合 / CEP / 多流关联)
                                    ↓
                           Kafka (DWS 汇总流)
                                    ↓
                      Flink Sink → ClickHouse (ADS)
                                    ↓
                              业务查询 / BI 看板

// Flink 1.16 Java API
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);

// 1. 从 Kafka 消费 ODS
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("ods_order")
    .setGroupId("flink_dwd_group")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> odsStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

// 2. 解析 JSON 并过滤（使用 Jackson）
SingleOutputStreamOperator<OrderEvent> orderStream = odsStream
    .map(new MapFunction<String, OrderEvent>() {
        @Override
        public OrderEvent map(String value) throws Exception {
            ObjectMapper mapper = new ObjectMapper();
            JsonNode node = mapper.readTree(value);
            if (node.get("order_id") == null || node.get("user_id") == null) {
                return null; // 脏数据
            }
            return new OrderEvent(
                node.get("order_id").asLong(),
                node.get("user_id").asLong(),
                node.get("product_id").asLong(),
                node.get("amount").asDouble(),
                node.get("create_time").asLong()
            );
        }
    })
    .filter(Objects::nonNull)
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getCreateTime())
    );

// 3. 异步关联维表（商品信息缓存）
DataStream<EnrichedOrder> enrichedStream = AsyncDataStream.unorderedWait(
    orderStream,
    new AsyncDimJoinFunction(),  // 实现 RichAsyncFunction，从 Redis/MySQL 查询
    1000, TimeUnit.MILLISECONDS,
    100  // 最大并发请求数
);

// 4. 写出到 Kafka DWD
KafkaSink<String> dwdSink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("dwd_order_enriched")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build();

enrichedStream.map(EnrichedOrder::toJson).sinkTo(dwdSink);

-- 使用 Flink SQL（生产环境可混合 Table API 做更复杂的逻辑）
CREATE TABLE dwd_order (
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    amount DOUBLE,
    create_time TIMESTAMP(3),
    product_name STRING,
    category STRING,
    WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'dwd_order_enriched',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json'
);

CREATE TABLE dws_product_minute (
    product_id BIGINT,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    total_amount DOUBLE,
    order_count BIGINT,
    PRIMARY KEY (product_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'dws_product_minute',
    'format' = 'json'
);

-- 开启 MiniBatch 优化（Flink 1.15+ 默认开启，但可显式设置）
SET table.exec.mini-batch.enabled = true;
SET table.exec.mini-batch.size = 5000;
SET table.exec.mini-batch.allow-latency = '5s';

INSERT INTO dws_product_minute
SELECT 
    product_id,
    TUMBLE_START(create_time, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(create_time, INTERVAL '1' MINUTE) AS window_end,
    SUM(amount) AS total_amount,
    COUNT(*) AS order_count
FROM dwd_order
GROUP BY product_id, TUMBLE(create_time, INTERVAL '1' MINUTE);

CREATE TABLE dws_product_minute_local ON CLUSTER clickhouse_cluster
(
    product_id UInt64,
    window_start DateTime,
    window_end DateTime,
    total_amount Float64,
    order_count UInt64,
    update_time DateTime DEFAULT now()
)
ENGINE = ReplacingMergeTree(update_time)
PARTITION BY toYYYYMMDD(window_start)
ORDER BY (product_id, window_start)
SETTINGS index_granularity = 8192;

-- 分布式表（查询用）
CREATE TABLE dws_product_minute_dist ON CLUSTER clickhouse_cluster
AS dws_product_minute_local
ENGINE = Distributed(clickhouse_cluster, default, dws_product_minute_local, rand());

public class ClickHouseBatchSink extends RichSinkFunction<DwsResult> {
    private transient ClickHouseConnection conn;
    private transient PreparedStatement stmt;
    private final List<DwsResult> buffer = new ArrayList<>();
    private static final int BATCH_SIZE = 10000;
    private static final int FLUSH_INTERVAL_MS = 5000;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.clickhouse.jdbc.ClickHouseDriver");
        String url = "jdbc:clickhouse://clickhouse-host:8123/default?socket_timeout=120000&connect_timeout=10000&async_insert=1&wait_for_async_insert=0";
        conn = DriverManager.getConnection(url);
        String sql = "INSERT INTO dws_product_minute_dist (product_id, window_start, window_end, total_amount, order_count) VALUES (?, ?, ?, ?, ?)";
        stmt = conn.prepareStatement(sql);
    }

    @Override
    public void invoke(DwsResult value, Context context) throws Exception {
        buffer.add(value);
        if (buffer.size() >= BATCH_SIZE) {
            flush();
        }
    }

    private void flush() throws SQLException {
        if (buffer.isEmpty()) return;
        for (DwsResult r : buffer) {
            stmt.setLong(1, r.productId);
            stmt.setTimestamp(2, new Timestamp(r.windowStart));
            stmt.setTimestamp(3, new Timestamp(r.windowEnd));
            stmt.setDouble(4, r.totalAmount);
            stmt.setLong(5, r.orderCount);
            stmt.addBatch();
        }
        stmt.executeBatch();
        buffer.clear();
    }

    @Override
    public void close() throws Exception {
        flush();
        if (stmt != null) stmt.close();
        if (conn != null) conn.close();
    }
}

-- 创建物化视图，按分钟预聚合
CREATE MATERIALIZED VIEW mv_product_minute_agg
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMMDD(window_start)
ORDER BY (product_id, window_start)
AS SELECT
    product_id,
    window_start,
    SUM(total_amount) AS sum_amount,
    SUM(order_count) AS sum_count
FROM dws_product_minute_dist
GROUP BY product_id, window_start;

-- 添加跳数索引加速过滤
ALTER TABLE mv_product_minute_agg ADD INDEX idx_amount sum_amount TYPE minmax GRANULARITY 2;

SELECT 
    product_id,
    sum(sum_amount) AS total,
    argMax(sum_count, window_start) AS last_count
FROM mv_product_minute_agg
WHERE window_start > now() - INTERVAL 5 MINUTE
GROUP BY product_id
ORDER BY total DESC
LIMIT 10
SETTINGS max_threads = 8, optimize_aggregation_in_order = 1;