可观测性层

可观测性的三大支柱各司其职：指标（Metrics）回答"现在怎么样"——低成本的数值时序，适合告警与趋势判断；日志（Logs）回答"为什么这样"——故障时刻的现场细节，适合根因定位；看板（Dashboards）把两者组织成"从全局到单点"的下钻路径，是值班同学的第一入口。湖仓链路长（采集 → 入湖 → 清洗 → 聚合 → 服务），任何一环都可能成为故障源，因此观测体系必须分层设计、逐层下钻，而不是堆一堆零散面板。

指标体系设计

指标不是越多越好，而是每个告警都必须有人在等。湖仓指标体系建议按四层组织：上层回答业务问题，下层支撑技术归因，下层异常要通过上层体现价值，上层异常要能下钻到下层定位根因。

Prometheus

1. 架构与 exporter 生态

Prometheus 采用拉取（Pull）模型：Server 周期性抓取各 exporter 暴露的 /metrics 端点，本地 TSDB 存储，经 PromQL 查询；告警规则在 Server 端评估，触发后推给 Alertmanager 分发。湖仓集群的 exporter 布局：

2. PromQL 常用查询

# 1) Kafka 消费积压：分区最新 offset − 消费组已提交 offset
sum by (consumergroup, topic) (kafka_consumergroup_group_lag)

# 2) 消费速率（条/秒），用于估算积压追平时间 = lag ÷ rate
sum by (topic) (rate(kafka_consumergroup_group_current_offset[5m]))

# 3) 调度作业失败率：30 分钟窗口失败任务占比
sum(rate(ds_task_failed_total[30m])) / clamp_min(sum(rate(ds_task_total[30m])), 1)

# 4) YARN 离线队列内存使用率（RM QueueMetrics 经 JMX 暴露，指标名以实际输出为准）
yarn_queue_allocated_mb{queue="offline"} / yarn_queue_total_mb{queue="offline"}

# 5) HDFS NameNode 5 分钟内 Full GC 累计秒数
increase(jvm_gc_collection_seconds_sum{role="namenode", gc="G1 Old Generation"}[5m])

# 6) 节点磁盘 IO 利用率（device 正则按实际盘名调整）
rate(node_disk_io_time_seconds_total{device=~"sd.*|nvme.*"}[5m])

3. Alertmanager：分组、抑制、静默

告警规则（Prometheus 侧）示例——覆盖组件层与作业/业务层两个规则组：

# prometheus/rules/lakehouse-alerts.yml
groups:
  - name: lakehouse.component          # 组件层规则组
    rules:
      - alert: NodeCpuHigh
        expr: (100 - avg by (instance) (rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100) > 85
        for: 5m
        labels:
          severity: warning
          layer: infra
        annotations:
          summary: "节点 CPU 持续偏高：{{ $labels.instance }}"
          runbook: "http://wiki.example/runbook/node-cpu"

      - alert: KafkaConsumerLagHigh
        expr: sum by (consumergroup, topic) (kafka_consumergroup_group_lag) > 100000
        for: 10m
        labels:
          severity: warning
          layer: component
        annotations:
          summary: "消费组 {{ $labels.consumergroup }} 在 {{ $labels.topic }} 积压 {{ $value }} 条"

      - alert: NameNodeGcPause
        expr: increase(jvm_gc_collection_seconds_sum{role="namenode", gc="G1 Old Generation"}[5m]) > 3
        labels:
          severity: critical
          layer: component
        annotations:
          summary: "HDFS NameNode 5 分钟 Full GC 累计 {{ $value }}s，存在全局卡顿风险"

  - name: lakehouse.job-biz            # 作业与业务层规则组
    rules:
      - alert: BatchJobFailureRate
        expr: |
          sum(rate(ds_task_failed_total[30m]))
            / clamp_min(sum(rate(ds_task_total[30m])), 1) > 0.05
        for: 15m
        labels:
          severity: critical
          layer: job
        annotations:
          summary: "调度作业 30 分钟失败率超过 5%"

      - alert: IngestDelayHigh
        expr: max(lakehouse_ingest_lag_seconds) > 900
        for: 10m
        labels:
          severity: warning
          layer: business
        annotations:
          summary: "入湖延迟超过 15 分钟，请下钻查看同步作业状态"

Alertmanager 侧路由配置——分组、抑制、静默三件套：

# alertmanager.yml
route:
  receiver: team-default
  group_by: ['alertname', 'cluster']     # 同名告警打包发送，减少轰炸
  group_wait: 30s                        # 组内等待归并的时间窗
  group_interval: 5m                     # 同组两次通知的最小间隔
  repeat_interval: 4h                    # 未恢复时的重复提醒周期
  routes:
    - matchers: ['severity="critical"']  # 严重级：升级到值班电话
      receiver: oncall-phone
      continue: true
    - matchers: ['team="stream"']        # 流团队告警路由到专属群
      receiver: stream-wecom

inhibit_rules:                           # 抑制：critical 触发时压掉同源 warning
  - source_matchers: ['severity="critical"']
    target_matchers: ['severity="warning"']
    equal: ['cluster', 'alertname']

receivers:
  - name: team-default
    webhook_configs:
      - url: 'http://wecom-proxy/alert/default'
  - name: oncall-phone
    webhook_configs:
      - url: 'http://pager-adapter/critical'
  - name: stream-wecom
    webhook_configs:
      - url: 'http://wecom-proxy/alert/stream'

# 静默（Silence）：维护窗口内按匹配器屏蔽告警，事由可审计
amtool silence add \
  --comment "node-03 换盘维护，CPU/磁盘告警预期内" \
  --duration 4h \
  alertname=~"NodeCpu.*|NodeDisk.*" instance=node-03

# 查看与过期静默规则
amtool silence ls
amtool silence expire <silence-id>

Grafana

1. 看板分层设计

湖仓看板的最大反模式是"一个大盘塞进一切"。正确姿势是三层漏斗，上层回答"有没有事"，下层回答"是哪里的事"：

层与层之间用面板 Data Link 串联：L1 的队列面板点击跳转到 L2 大盘并自动带入 $queue 变量；L2 的作业面板跳转 L3 并带入 $job_id；L3 面板携带时间范围直链 Kibana Discover。值班同学从收到告警到看到日志现场，全程不需要手动改查询条件。

2. 变量与面板设计

EFK 日志链路

1. 全链路拓扑

EFK = Filebeat（采集）+ Elasticsearch（存储与检索）+ Kibana（查询与可视化）。湖仓集群日志量往往在百 GB/天量级，直接把 Filebeat 打到 ES 会在重写/恢复期形成反压，因此推荐引入 Kafka 作为缓冲层削峰：

2. filebeat.yml 完整配置

三个关键设计：多来源分 input（每类组件一个 input，便于差异化处理）；multiline 合并 Java stacktrace（不以时间戳开头的行为同一条日志）；fields 打标（cluster/service/role 贯穿到 Kibana 检索与 ILM）。

# ---------- 1) HDFS NameNode：JVM 长堆栈需要 multiline 合并 ----------
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/hadoop/*namenode*.log
    fields:
      cluster: prod-lakehouse
      service: hdfs
      role: namenode
    fields_under_root: true        # 字段平铺到根，KQL 直接用 service: hdfs
    multiline.pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'   # 以时间戳开头才算新日志行
    multiline.negate: true
    multiline.match: after
    multiline.max_lines: 200       # 单条日志合并上限，防畸形日志撑爆事件
    tags: ["hdfs"]

  # ---------- 2) Flink TaskManager：同样需要 stacktrace 合并 ----------
  - type: log
    enabled: true
    paths:
      - /var/log/flink/taskmanager*.log
    fields:
      cluster: prod-lakehouse
      service: flink
      role: taskmanager
    fields_under_root: true
    multiline.pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'
    multiline.negate: true
    multiline.match: after
    tags: ["flink"]

  # ---------- 3) 调度与同步组件日志 ----------
  - type: log
    enabled: true
    paths:
      - /var/log/dolphinscheduler/*.log
      - /var/log/datax/*.log
    fields:
      cluster: prod-lakehouse
      service: scheduler
      role: worker
    fields_under_root: true
    tags: ["scheduler"]

processors:
  - add_host_metadata: ~          # 自动补充 host.name / host.ip
  - add_fields:
      target: ''
      fields:
        env: prod

# ---------- 输出：大规模走 Kafka 缓冲 ----------
output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: "log_lakehouse_all"
  required_acks: 1                 # 性能与可靠性的折中
  compression: gzip
  worker: 2

# ---------- 小规模直写 ES：注释掉 Kafka 输出后启用 ----------
# output.elasticsearch:
#   hosts: ["http://es-lb:9200"]
#   index: "logs-lakehouse-%{+yyyy.MM.dd}"
#   bulk_max_size: 4096

3. ES ILM 索引生命周期

湖仓日志的访问特征是前 3 天高频检索、之后几乎不再访问，用 ILM 按热/温/冷/删除四阶段滚动，比手动 curator 定时任务可靠得多：

PUT _ilm/policy/lakehouse-logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": { "max_primary_shard_size": "50gb", "max_age": "1d" },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "3d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },          // 合并段，检索量骤降后压缩存储
          "allocate": { "number_of_replicas": 0,
                        "require": { "data": "warm" } }
        }
      },
      "cold": {
        "min_age": "15d",
        "actions": {
          "set_priority": { "priority": 50 },
          "allocate": { "require": { "data": "cold" } }      // 开源版回退方案：挪到冷存储节点
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": { "delete": {} }                        // 保留 30 天，审计诉求可延长 cold 阶段
      }
    }
  }
}

# 绑定到索引模板（新建索引自动应用）
PUT _index_template/logs-lakehouse
{
  "index_patterns": ["logs-lakehouse-*"],
  "template": {
    "settings": {
      "index.lifecycle.name": "lakehouse-logs-policy",
      "index.lifecycle.rollover_alias": "logs-lakehouse",   // rollover 需要别名配合
      "number_of_shards": 3,
      "number_of_replicas": 1
    }
  }
}

4. Kibana 检索用法

Kibana Discover 使用 KQL（Kibana Query Language）做前缀式检索，配合 filebeat 打的 fields，常用姿势：

# 1) 定位 Flink Checkpoint 相关的 ERROR 日志
service: "flink" and role: "taskmanager" and level: "ERROR" and message: *checkpoint*

# 2) 查被 YARN 杀掉的 Container（定位内存超限）
service: "yarn" and message: "Container killed on request" and cluster: "prod-lakehouse"

# 3) 按服务聚合 ERROR 数量：KQL 过滤后，用 Visualize
#    按 service 做 Terms 聚合、按 @timestamp 做 Date histogram（每 5 分钟）
service: "spark" and level: "ERROR"

湖仓特有监控点

通用大数据监控覆盖不到湖仓的"慢性病"：小文件、快照、Compaction 这些问题不报警、也不崩溃，只会让性能一寸寸劣化。以下五个监控点是湖仓集群的"体检必查项"：

湖仓表健康度类指标没有现成 exporter，用 SQL exporter 思路——定时执行元数据查询，把结果暴露为 Prometheus 指标：

-- SQL exporter 定时执行：把湖仓表健康度变成指标
-- 1) 快照堆积（对应 iceberg_snapshot_total 指标）
SELECT count(*) AS iceberg_snapshot_total FROM lakehouse.dws_orders_day.snapshots;

-- 2) 数据文件数（小文件规模，对应 iceberg_file_num）
SELECT count(*) AS iceberg_file_num FROM lakehouse.dws_orders_day.files;

-- 3) 数据新鲜度：最新分区距当前的小时数（对应 lakehouse_freshness_hours）
SELECT round((unix_timestamp() - unix_timestamp(max(dt), 'yyyy-MM-dd')) / 3600)
       AS lakehouse_freshness_hours
FROM lakehouse.dws_orders_day;

# Flink Checkpoint 告警规则（Prometheus 规则片段）
# 时长超过 60s 预示背压或状态膨胀（间隔按流作业实际 checkpoint 配置调整）
flink_jobmanager_job_lastCheckpointDuration > 60000

# 连续失败：5 分钟窗口内失败检查点增量 ≥ 2 次即为 Critical
increase(flink_jobmanager_job_numberOfFailedCheckpoints[5m]) >= 2

调优要点与边界

1. 采样率与基数爆炸

Prometheus 最常见的死法不是抓不动，而是时间序列基数爆炸——Spark/Flink 每个作业、每个 executor、每个算子都会生成带唯一标签的序列，高峰期序列数轻松破千万。治理手段：

# 采集频率分级：基础设施 30s，组件 60s，业务指标 60s
global:
  scrape_interval: 30s
  scrape_timeout: 10s

scrape_configs:
  - job_name: 'spark-pushgateway'
    honor_labels: true
    scrape_interval: 60s
    metric_relabel_configs:
      # 1) 按 metric 名丢弃高基数序列（流式 gauge 常见每秒一变）
      - source_labels: [__name__]
        regex: 'spark_.+_gauges_.+|spark_.+_live_.+'
        action: drop
      # 2) 剔除高基数标签，保留聚合维度
      - action: labeldrop
        regex: '(executor_id|attempt_id|application_id)'

• 容量参考：单实例活跃序列建议控制在 50 万以内，全集群控制在 5000 万以内；超限时优先 drop，其次才考虑分片（Thanos / VictoriaMetrics）；
• 抓取超时应小于抓取间隔的一半，并监控 up == 0、scrape_duration_seconds > interval × 0.8——监控系统需要自己的监控。

2. 日志分级采集

照单全收的日志采集会把 70% 的 ES 存储浪费在没人看的 INFO 上。在 Filebeat 端（越早丢弃成本越低）分级：

# 日志分级：ERROR/FATAL 全量，DEBUG 直接丢弃，高噪服务的 INFO 也裁掉
processors:
  - drop_event.when:
      equals:
        level: "DEBUG"
  - drop_event.when:
      and:
        - equals:
            level: "INFO"
        - equals:
            service: "yarn-nodemanager"     # 心跳类 INFO 日志量大且无价值

• ERROR / FATAL：全量采集，全链路可检索；
• WARN：全量采集，但 ILM 缩短其 warm 停留（价值衰减快）；
• INFO：仅保留服务边界事件（启停、拓扑变更），高噪服务直接在采集端丢弃。

3. 何时用 Loki 替代 ES