智慧供热湖仓一体化落地指南：存储、入湖、查询三层拆解

传感器/仪表 → 采集网关(PLC/RTU) → SCADA/前置机 → Kafka → Flink → Iceberg

{
  "station_id": "ST-2024-001",
  "device_id": "DEV-PLC-001",
  "metrics": {
    "primary_supply_temp": 85.2,
    "primary_return_temp": 60.1,
    "secondary_supply_temp": 50.3,
    "secondary_return_temp": 43.7,
    "primary_pressure": 0.82,
    "primary_flow": 142.5,
    "secondary_flow": 88.3,
    "heat_power": 1850.6,
    "circulation_pump_freq": 35.2,
    "makeup_pump_status": 1,
    "valve_opening": 65
  },
  "quality": 192,
  "event_time": "2026-01-15T08:30:00.000+08:00",
  "ingest_time": "2026-01-15T08:30:01.200+08:00"
}

minio server /data{1...8} \
  --console-address ":9001" \
  --address ":9000"
# 环境变量
MINIO_ROOT_USER=heat_admin
MINIO_ROOT_PASSWORD=<强密码>
MINIO_DOMAIN=minio.heat.local   # 支持虚拟主机风格寻址

s3://warehouse/heat_catalog/ods/scada_metric_raw/
├── data/
│   ├── event_time_day=2026-01-15/
│   │   ├── station_id_bucket=0/
│   │   │   ├── 00000-0-1a2b3c4d-5e6f-7a8b-9c0d.parquet      ← 数据文件
│   │   │   ├── 00001-0-2b3c4d5e-6f7a-8b9c-0d1e.parquet
│   │   │   └── 00002-0-3c4d5e6f-7a8b-9b0c-1d2e.parquet
│   │   ├── station_id_bucket=1/
│   │   │   └── 00000-0-4d5e6f7a-8b9c-0d1e-2f3a.parquet
│   │   ├── ... (station_id_bucket=2 ~ 15)
│   │   └── station_id_bucket=15/
│   │       └── 00000-0-5e6f7a8b-9c0d-1e2f-3a4b.parquet
│   ├── event_time_day=2026-01-16/
│   │   └── station_id_bucket=0/
│   │       └── ...
│   └── ...
├── metadata/
│   ├── v1.metadata.json          ← 表初始 metadata
│   ├── v2.metadata.json          ← 第一次写入后
│   ├── v3.metadata.json          ← 第二次写入后
│   ├── snap-1640995200000-1.avro ← manifest list（快照入口）
│   ├── snap-1640995260000-2.avro
│   ├── 1a2b3c4d-5e6f-7a8b-9c0d.m0.avro  ← manifest 文件
│   ├── 2b3c4d5e-6f7a-8b9c-0d1e.m0.avro
│   └── ...

Parquet 文件
├── Row Group 1 (128MB，约 100 万行)
│   ├── Column Chunk: station_id (STRING)
│   │   ├── Page 1 (字典编码 + RLE，压缩后 ~50KB)
│   │   └── Page 2
│   ├── Column Chunk: metric_code (STRING)
│   │   └── Page 1 (字典编码，指标码就十几种，压缩率极高)
│   ├── Column Chunk: metric_value (DOUBLE)
│   │   └── Page 1 (温度值 85.2/85.3/85.1...，Delta 编码 + zstd，压缩率 ~10x)
│   └── Column Chunk: event_time (TIMESTAMP)
│       └── Page 1 (时间戳递增，Delta 编码，压缩率 ~20x)
├── Row Group 2
└── Footer (schema + column statistics: min/max/null_count)

# Iceberg REST Catalog 服务环境变量
AWS_ACCESS_KEY_ID=heat_admin
AWS_SECRET_ACCESS_KEY=<强密码>
AWS_REGION=us-east-1
CATALOG_WAREHOUSE=s3://warehouse/
CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
CATALOG_S3_ENDPOINT=http://minio:9000
CATALOG_S3_PATH_STYLE_ACCESS=true

<!-- Flink 核心（按版本选，以 1.19 为例）-->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.12</artifactId>
  <version>1.19.1</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.12</artifactId>
  <version>3.2.0-1.19</version>
</dependency>
<!-- Iceberg Flink runtime -->
<dependency>
  <groupId>org.apache.iceberg</groupId>
  <artifactId>iceberg-flink-runtime-1.19</artifactId>
  <version>1.5.0</version> <!-- 或 1.6.x，与 Doris SDK 对齐 -->
</dependency>
<!-- Iceberg AWS S3（S3FileIO 依赖）-->
<dependency>
  <groupId>org.apache.iceberg</groupId>
  <artifactId>iceberg-aws-bundle</artifactId>
  <version>1.5.0</version>
</dependency>

// 1. 配置 Iceberg Catalog Loader
Configuration icebergConf = new Configuration();
icebergConf.set(CatalogProperties.CATALOG_IMPL, "org.apache.iceberg.rest.RESTCatalog");
icebergConf.set(CatalogProperties.URI, "http://rest:8181");
icebergConf.set(CatalogProperties.WAREHOUSE_LOCATION, "s3://warehouse/");
icebergConf.set("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
icebergConf.set("s3.endpoint", "http://minio:9000");
icebergConf.set("s3.access-key-id", "heat_admin");
icebergConf.set("s3.secret-access-key", "<密码>");
icebergConf.set("s3.path-style-access", "true");

CatalogLoader catalogLoader = CatalogLoader.custom(
    "heat_catalog", icebergConf.toMap(),
    new HashMap<>(), "org.apache.iceberg.rest.RESTCatalog");

TableLoader tableLoader = TableLoader.fromCatalog(catalogLoader,
    TableIdentifier.of("ods", "scada_metric_raw"));

// 2. Kafka Source
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("heat.scada.station_metric")
    .setGroupId("flink-scada-to-iceberg")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

// 3. 解析 + 清洗
DataStream<RowData> cleanedStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source")
    .map(new ScadaParseAndCleanMap())   // JSON → RowData + 清洗
    .filter(Objects::nonNull)           // 过滤掉清洗失败的脏数据
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<RowData>forBoundedOutOfOrderness(Duration.ofSeconds(30))
            .withTimestampAssigner((row, ts) ->
                row.getTimestamp(6, 6).getMillisecond())); // event_time 字段索引 6

public class ScadaParseAndCleanMap extends MapFunction<String, RowData> {
    // 物理量合理范围（用于去野值）
    static final double TEMP_MIN = -30.0, TEMP_MAX = 150.0;
    static final double PRESSURE_MIN = 0.0, PRESSURE_MAX = 2.5;
    static final double FLOW_MIN = 0.0, FLOW_MAX = 5000.0;

    @Override
    public RowData map(String json) throws Exception {
        JSONObject obj = JSON.parseObject(json);
        String stationId = obj.getString("station_id");
        JSONObject m = obj.getJSONObject("metrics");
        long eventTimeMs = parseToMillis(obj.getString("event_time"));

        // 去野值：超出物理量程的置 NULL，不丢弃，保留记录便于追溯
        Double supplyTemp = sanitize(m.getDouble("primary_supply_temp"), TEMP_MIN, TEMP_MAX);
        Double pressure  = sanitize(m.getDouble("primary_pressure"), PRESSURE_MIN, PRESSURE_MAX);
        Double flow      = sanitize(m.getDouble("primary_flow"), FLOW_MIN, FLOW_MAX);

        // 构造 RowData（字段顺序与 Iceberg 表 schema 对齐）
        GenericRowData row = new GenericRowData(8);
        row.setField(0, StringData.fromString(stationId));
        row.setField(1, StringData.fromString(obj.getString("device_id")));
        row.setField(2, StringData.fromString("PRIMARY_SUPPLY_TEMP"));
        row.setField(3, supplyTemp);  // null 表示野值
        row.setField(4, obj.getIntValue("quality"));
        row.setField(5, TimestampData.fromEpochMillis(eventTimeMs));
        row.setField(6, TimestampData.fromEpochMillis(System.currentTimeMillis()));
        return row;
    }

    private Double sanitize(Double val, double min, double max) {
        if (val == null) return null;
        return (val >= min && val <= max) ? val : null; // 超量程置 NULL
    }
}

WatermarkStrategy.<RowData>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((row, ts) ->
        row.getTimestamp(5, 6).getMillisecond()) // event_time 字段
    .withIdleness(Duration.ofMinutes(1));         // 防止空闲分区阻塞水位线

// 4. 写入 Iceberg
FlinkSink.forRowData(cleanedStream)
    .tableLoader(tableLoader)
    .set("write.target-file-size-bytes", String.valueOf(128 * 1024 * 1024)) // 128MB
    .set("write.distribution-mode", "hash")   // 按分区键 hash 分布到各 task
    .set("write.parquet.compression-codec", "zstd")
    .set("write.metadata.delete-after-commit.enabled", "true") // 提交后删旧 metadata
    .set("write.metadata.metrics.default", "full")             // 记录完整列统计
    .append();  // 追加模式，不覆盖

// 5. Checkpoint 配置
env.enableCheckpointing(5 * 60 * 1000); // 5 分钟一次
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000); // 10 分钟超时
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60 * 1000); // 两次 checkpoint 间至少 1 分钟
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 作业取消时保留 checkpoint

// 6. State Backend（RocksDB，支持大状态增量 checkpoint）
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///opt/flink/checkpoints/");

// 在 Flink 作业中定期触发 compaction（只针对冷分区）
Actions.forTable(table)
    .rewriteDataFiles()
    .targetFileSizeInBytes(128 * 1024 * 1024)
    .filter(Expressions.lessThan("event_time",
        currentHourMinus2Hours())) // 只 compact 2 小时前的冷分区
    .execute();

-- 创建 Kafka 源表
CREATE TABLE kafka_scada_source (
    station_id STRING,
    device_id STRING,
    metrics ROW<
        primary_supply_temp DOUBLE,
        primary_return_temp DOUBLE,
        primary_pressure DOUBLE,
        primary_flow DOUBLE,
        heat_power DOUBLE
    >,
    quality INT,
    event_time TIMESTAMP(6),
    ingest_time TIMESTAMP(6),
    WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'heat.scada.station_metric',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink-scada-sql',
    'scan.startup.mode' = 'group-offsets',
    'format' = 'json'
);

-- 创建 Iceberg 目标表（通过 REST Catalog）
CREATE CATALOG heat_catalog WITH (
    'type' = 'iceberg',
    'catalog-impl' = 'org.apache.iceberg.rest.RESTCatalog',
    'uri' = 'http://rest:8181',
    'warehouse' = 's3://warehouse/',
    'io-impl' = 'org.apache.iceberg.aws.s3.S3FileIO',
    's3.endpoint' = 'http://minio:9000',
    's3.access-key-id' = 'heat_admin',
    's3.secret-access-key' = '<密码>',
    's3.path-style-access' = 'true'
);

-- 展平 metrics 并写入 Iceberg
INSERT INTO heat_catalog.ods.scada_metric_flat
SELECT
    station_id, device_id,
    metrics.primary_supply_temp AS supply_temp,
    metrics.primary_return_temp  AS return_temp,
    metrics.primary_pressure     AS pressure,
    metrics.primary_flow         AS flow,
    metrics.heat_power           AS heat_power,
    quality, event_time, ingest_time
FROM kafka_scada_source
WHERE metrics.primary_supply_temp BETWEEN -30 AND 150  -- 去野值
  AND metrics.primary_pressure BETWEEN 0 AND 2.5;

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("heat-lakehouse-etl") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.heat_catalog",
            "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.heat_catalog.type", "rest") \
    .config("spark.sql.catalog.heat_catalog.uri", "http://rest:8181") \
    .config("spark.sql.catalog.heat_catalog.warehouse", "s3://warehouse/") \
    .config("spark.sql.catalog.heat_catalog.io-impl",
            "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.heat_catalog.s3.endpoint",
            "http://minio:9000") \
    .config("spark.sql.catalog.heat_catalog.s3.access-key-id", "heat_admin") \
    .config("spark.sql.catalog.heat_catalog.s3.secret-access-key", "<密码>") \
    .config("spark.sql.catalog.heat_catalog.s3.path-style-access", "true") \
    .config("spark.sql.defaultCatalog", "heat_catalog") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

<!-- Spark 3.5 + Scala 2.12 -->
<dependency>
  <groupId>org.apache.iceberg</groupId>
  <artifactId>iceberg-spark-runtime-3.5_2.12</artifactId>
  <version>1.5.0</version>
</dependency>
<dependency>
  <groupId>org.apache.iceberg</groupId>
  <artifactId>iceberg-aws-bundle</artifactId>
  <version>1.5.0</version>
</dependency>

-- DWD: 换热站运行数据宽表（按站×分钟粒度）
INSERT INTO heat_catalog.dwd.station_metric_wide
SELECT
    station_id,
    device_id,
    event_time,
    -- 透视：把 metric_code 行转列
    MAX(CASE WHEN metric_code = 'PRIMARY_SUPPLY_TEMP' THEN metric_value END) AS primary_supply_temp,
    MAX(CASE WHEN metric_code = 'PRIMARY_RETURN_TEMP'  THEN metric_value END) AS primary_return_temp,
    MAX(CASE WHEN metric_code = 'SECONDARY_SUPPLY_TEMP' THEN metric_value END) AS secondary_supply_temp,
    MAX(CASE WHEN metric_code = 'SECONDARY_RETURN_TEMP' THEN metric_value END) AS secondary_return_temp,
    MAX(CASE WHEN metric_code = 'PRIMARY_PRESSURE' THEN metric_value END) AS primary_pressure,
    MAX(CASE WHEN metric_code = 'PRIMARY_FLOW' THEN metric_value END) AS primary_flow,
    MAX(CASE WHEN metric_code = 'SECONDARY_FLOW' THEN metric_value END) AS secondary_flow,
    MAX(CASE WHEN metric_code = 'HEAT_POWER' THEN metric_value END) AS heat_power,
    MAX(CASE WHEN metric_code = 'VALVE_OPENING' THEN metric_value END) AS valve_opening,
    -- 供回水温差（衍生指标）
    MAX(CASE WHEN metric_code = 'PRIMARY_SUPPLY_TEMP' THEN metric_value END)
      - MAX(CASE WHEN metric_code = 'PRIMARY_RETURN_TEMP' THEN metric_value END) AS primary_temp_diff,
    -- 数据质量：任一关键指标为 NULL 则标记 0
    CASE WHEN
      MAX(CASE WHEN metric_code = 'PRIMARY_SUPPLY_TEMP' THEN metric_value END) IS NULL
      OR MAX(CASE WHEN metric_code = 'PRIMARY_FLOW' THEN metric_value END) IS NULL
      THEN 0 ELSE 1
    END AS data_quality_flag,
    current_timestamp() AS etl_time
FROM heat_catalog.ods.scada_metric_raw
WHERE date(event_time) = date_sub(current_date(), 1)  -- 处理前一天数据
GROUP BY station_id, device_id, event_time;

-- DWS: 换热站能耗日表
INSERT INTO heat_catalog.dws.station_energy_daily
SELECT
    station_id,
    date(event_time) AS stat_date,
    -- 温度统计
    avg(primary_supply_temp)  AS avg_supply_temp,
    avg(primary_return_temp)  AS avg_return_temp,
    avg(primary_temp_diff)    AS avg_temp_diff,
    min(primary_supply_temp)  AS min_supply_temp,
    max(primary_supply_temp)  AS max_supply_temp,
    -- 流量与热量
    avg(primary_flow)         AS avg_flow,
    sum(primary_flow * 60)    AS daily_flow_volume,  -- L/min × 60min = L/h → 累计
    sum(heat_power * 60 / 1e6) AS daily_heat_gj,      -- W × 60s / 1e6 = GJ
    -- 运行时长（有数据的小时数）
    count(DISTINCT hour(event_time)) AS running_hours,
    -- 数据完整率（有效记录数 / 应有记录数 1440）
    count(*) / 1440.0 AS data_completeness
FROM heat_catalog.dwd.station_metric_wide
WHERE date(event_time) = date_sub(current_date(), 1)
  AND data_quality_flag = 1
GROUP BY station_id, date(event_time);

-- ADS: 室温达标率（室温 ≥ 18℃ 视为达标）
INSERT INTO heat_catalog.ads.room_temp_compliance
SELECT
    date(event_time) AS stat_date,
    station_id,
    count(*) AS total_records,
    count(CASE WHEN room_temp >= 18.0 THEN 1 END) AS compliant_records,
    round(count(CASE WHEN room_temp >= 18.0 THEN 1 END) * 100.0 / count(*), 2) AS compliance_rate,
    avg(room_temp) AS avg_room_temp
FROM heat_catalog.ods.room_temp_raw
WHERE date(event_time) = date_sub(current_date(), 1)
GROUP BY date(event_time), station_id;

-- bin-pack 合并（默认策略，速度快）
CALL system.rewrite_data_files(
  table => 'heat_catalog.ods.scada_metric_raw',
  options => map(
    'target-file-size-bytes', '134217728',  -- 128MB
    'min-input-files', '5',                  -- 至少 5 个小文件才合并
    'min-file-size-bytes', '10485760'        -- 小于 10MB 视为小文件
  )
);

-- Z-order 重排（按 station_id + event_time 联合排序）
CALL system.rewrite_data_files(
  table => 'heat_catalog.ods.scada_metric_raw',
  options => map(
    'strategy', 'sort',
    'sort-order', 'zorder(station_id, event_time)',
    'target-file-size-bytes', '134217728'
  )
);

-- 过期 30 天前的快照（保留最近 30 天用于 Time Travel 回溯）
CALL system.expire_snapshots(
  table => 'heat_catalog.ods.scada_metric_raw',
  older_than => TIMESTAMP '2026-05-28 00:00:00',
  retain_last => 10  -- 至少保留最近 10 个快照
);

CALL system.remove_orphan_files(
  table => 'heat_catalog.ods.scada_metric_raw',
  older_than => TIMESTAMP '2026-06-26 00:00:00'  -- 只删 24 小时前的，避免删到正在写的
);

CALL system.rewrite_manifests('heat_catalog.ods.scada_metric_raw');

#!/bin/bash
# /opt/heat/jobs/daily_maintenance.sh — 每天凌晨 2 点执行
TODAY=$(date +%Y-%m-%d)
YESTERDAY=$(date -d "yesterday" +%Y-%m-%d)
THIRTY_DAYS_AGO=$(date -d "30 days ago" +%Y-%m-%d)

spark-sql --master yarn --name "iceberg-maintenance" <<EOF

-- 1. 合并前天及更早的冷分区小文件（Z-order）
CALL system.rewrite_data_files(
  table => 'heat_catalog.ods.scada_metric_raw',
  options => map('strategy','sort','sort-order','zorder(station_id, event_time)')
);

-- 2. DWS/DWD 层 bin-pack 合并
CALL system.rewrite_data_files('heat_catalog.dwd.station_metric_wide');
CALL system.rewrite_data_files('heat_catalog.dws.station_energy_daily');

-- 3. 过期 30 天前的快照
CALL system.expire_snapshots('heat_catalog.ods.scada_metric_raw',
  TIMESTAMP '${THIRTY_DAYS_AGO} 00:00:00', 10);

-- 4. 清理孤儿文件（24 小时前）
CALL system.remove_orphan_files(table => 'heat_catalog.ods.scada_metric_raw',
  older_than => TIMESTAMP '${YESTERDAY} 00:00:00');

-- 5. 合并 manifest
CALL system.rewrite_manifests('heat_catalog.ods.scada_metric_raw');
EOF

-- 查看快照历史
SELECT snapshot_id, committed_at, operation, summary
FROM heat_catalog.ods.scada_metric_raw.snapshots
ORDER BY committed_at DESC LIMIT 10;

-- 查看数据文件分布（识别小文件问题）
SELECT
    count(*) AS file_count,
    avg(file_size_in_bytes) AS avg_size_mb,
    max(file_size_in_bytes) AS max_size_mb,
    min(file_size_in_bytes) AS min_size_mb
FROM heat_catalog.ods.scada_metric_raw.files
WHERE partition.event_time_day = '${YESTERDAY}';

-- 查看分区数据量
SELECT partition.event_time_day, file_count, record_count, total_data_file_size_in_bytes
FROM heat_catalog.ods.scada_metric_raw.partitions
ORDER BY partition.event_time_day DESC LIMIT 7;

-- 创建 Iceberg Catalog（REST Catalog + MinIO）
CREATE CATALOG iceberg PROPERTIES (
    'type'                 = 'iceberg',
    'iceberg.catalog.type' = 'rest',
    'warehouse'            = 's3://warehouse/',
    'uri'                  = 'http://rest:8181',
    's3.access_key'        = 'heat_admin',
    's3.secret_key'        = '<密码>',
    's3.endpoint'          = 'http://minio:9000',
    's3.region'            = 'us-east-1'
);

-- 验证：列出 Iceberg 中的数据库和表
SWITCH iceberg;
SHOW DATABASES;  -- 应看到 ods, dwd, dws, ads, dim
USE ods;
SHOW TABLES;     -- 应看到 scada_metric_raw 等

-- 切回 internal catalog 做跨源查询
SWITCH internal;

-- 查询 Iceberg 湖中的 DWS 日表（直读，无加速）
SELECT station_id, stat_date, avg_supply_temp, daily_heat_gj
FROM iceberg.dws.station_energy_daily
WHERE stat_date = '2026-01-15'
ORDER BY daily_heat_gj DESC
LIMIT 20;

-- 物化视图：换热站实时温度（从 ODS 直读加速）
CREATE MATERIALIZED VIEW mv_station_realtime_temp
DISTRIBUTED BY HASH(station_id) BUCKETS 16
REFRESH ASYNC EVERY 1 MINUTE  -- 每分钟刷新
PROPERTIES (
    'replication_num' = '2'
)
AS
SELECT
    station_id,
    event_time,
    primary_supply_temp,
    primary_return_temp,
    primary_pressure,
    primary_flow
FROM iceberg.dwd.station_metric_wide
WHERE event_time >= now() - INTERVAL '1' HOUR;  -- 只缓存最近 1 小时

CREATE MATERIALIZED VIEW mv_energy_daily_topn
DISTRIBUTED BY HASH(stat_date) BUCKETS 3
REFRESH ASYNC EVERY 1 HOUR  -- 每小时刷新
PROPERTIES ('replication_num' = '2')
AS
SELECT
    stat_date,
    station_id,
    avg_supply_temp,
    daily_heat_gj,
    data_completeness,
    -- 单方能耗（GJ/㎡，需关联维度表）
    daily_heat_gj / area_heated AS unit_heat_consumption
FROM iceberg.dws.station_energy_daily d
JOIN internal.dim.dim_station s ON d.station_id = s.station_id
WHERE stat_date >= date_trunc(now(), 'month');  -- 本月数据

SELECT station_id, avg_supply_temp
FROM iceberg.dwd.station_metric_wide
WHERE event_time >= now() - INTERVAL '30' MINUTE;

-- 创建 MySQL Catalog（维度数据源）
CREATE CATALOG mysql_dim PROPERTIES (
    'type' = 'jdbc',
    'user' = 'readonly',
    'password' = '<密码>',
    'jdbc_url' = 'jdbc:mysql://mysql:3306/heat_dim',
    'driver_url' = 'mysql-connector-java-8.0.30.jar',
    'driver_class' = 'com.mysql.cj.jdbc.Driver'
);

-- 跨源 JOIN：湖中运行数据 + MySQL 用户档案
SELECT
    s.station_id,
    u.customer_name,
    u.building_name,
    s.avg_supply_temp,
    s.daily_heat_gj
FROM iceberg.dws.station_energy_daily s
JOIN mysql_dim.heat_customer u ON s.station_id = u.bind_station
WHERE s.stat_date = current_date()
ORDER BY s.daily_heat_gj DESC;

-- 在 Iceberg 中建表
CREATE TABLE iceberg.ads.load_forecast_result (
    forecast_date DATE,
    station_id STRING,
    predicted_load_gj DOUBLE,
    confidence DOUBLE,
    model_version STRING,
    forecast_time TIMESTAMP
) PARTITIONED BY (days(forecast_date));

-- 把 Doris 内部计算的结果写回 Iceberg
INSERT INTO iceberg.ads.load_forecast_result
SELECT
    date_add(current_date(), 1) AS forecast_date,
    station_id,
    predicted_load_gj,
    confidence,
    'v2.1' AS model_version,
    current_timestamp() AS forecast_time
FROM internal.ads.load_forecast_model_output
WHERE forecast_date = date_add(current_date(), 1);

CREATE TABLE heat_catalog.ods.scada_metric_raw (
    station_id     STRING,       -- 换热站编号
    device_id      STRING,       -- 设备编号
    metric_code    STRING,       -- 指标码: PRIMARY_SUPPLY_T/PRIMARY_RETURN_T/...
    metric_value   DOUBLE,       -- 采集值（野值置 NULL）
    quality        INT,          -- 数据质量位掩码
    event_time     TIMESTAMP(6), -- 采集时间(事件时间)
    ingest_time    TIMESTAMP(6)  -- 入湖时间
)
PARTITIONED BY (days(event_time), bucket(16, station_id))
TBLPROPERTIES (
    'write.format.default'            = 'parquet',
    'write.parquet.compression-codec' = 'zstd',
    'write.target-file-size-bytes'    = '134217728',  -- 128MB
    'write.metadata.metrics.default'  = 'full',        -- 完整列统计
    'format-version'                  = '2'            -- V2 表，支持行级删除
);

CREATE TABLE heat_catalog.dws.station_energy_daily (
    station_id          STRING,
    stat_date           DATE,
    avg_supply_temp     DOUBLE,
    avg_return_temp     DOUBLE,
    avg_temp_diff       DOUBLE,
    min_supply_temp     DOUBLE,
    max_supply_temp     DOUBLE,
    avg_flow            DOUBLE,
    daily_flow_volume   DOUBLE,   -- 日累计流量(L)
    daily_heat_gj       DOUBLE,   -- 日供热量(GJ)
    running_hours       INT,      -- 运行时长
    data_completeness   DOUBLE,   -- 数据完整率
    etl_time            TIMESTAMP(6)
)
PARTITIONED BY (days(stat_date))
TBLPROPERTIES (
    'write.parquet.compression-codec' = 'zstd',
    'write.target-file-size-bytes'    = '134217728',
    'format-version'                  = '2'
);

-- 历史数据增加月分区（新数据仍按天，历史数据按月共存）
ALTER TABLE heat_catalog.ods.scada_metric_raw
  ADD PARTITION FIELD months(event_time);

-- 删除 1 年前的 ODS 分区
ALTER TABLE heat_catalog.ods.scada_metric_raw
  DROP PARTITION FIELD days(event_time);
-- 或用 Spark 批量删除旧分区数据
DELETE FROM heat_catalog.ods.scada_metric_raw
WHERE event_time < TIMESTAMP '2025-01-01 00:00:00';