1. 故障排查体系框架

1.1 分层诊断模型

系统化的问题定位方法。

故障排查层次结构
├── 应用层故障
│   ├── SQL语法错误
│   ├── 业务逻辑错误
│   └── 数据质量异常
├── 运行时故障
│   ├── 资源不足
│   ├── 状态后端异常
│   └── 网络连接问题
├── 基础设施故障
│   ├── 集群节点宕机
│   ├── 存储系统异常
│   └── 网络分区故障
└── 数据源/目标故障
    ├── 连接器配置错误
    ├── 外部系统不可用
    └── 数据格式不兼容

1.2 排查工具链

完整的调试工具集合。

-- 1. 系统信息查询
SHOW TABLES;                    -- 查看已注册表
SHOW FUNCTIONS;                 -- 查看可用函数
DESCRIBE table_name;            -- 表结构信息
EXPLAIN PLAN FOR query;         -- 执行计划分析

-- 2. 运行时状态查询
SELECT * FROM table_name;       -- 数据采样检查
SELECT COUNT(*) FROM stream;    -- 数据量检查

-- 3. 性能指标监控
SELECT * FROM metrics_stream;   -- 自定义指标监控

2. SQL语法与语义错误排查

2.1 常见语法错误诊断

快速定位SQL编写问题。

-- 错误示例1: 字段名错误
-- 错误: SELECT user_nam FROM users; (字段不存在)
-- 正确: 
SELECT user_name FROM users;

-- 错误示例2: 类型不匹配
-- 错误: SELECT '100' + 200; (字符串与数字相加)
-- 正确:
SELECT CAST('100' AS INT) + 200;

-- 错误示例3: 窗口函数使用错误
-- 错误: SELECT ROW_NUMBER() OVER () FROM events; (缺少ORDER BY)
-- 正确:
SELECT ROW_NUMBER() OVER (ORDER BY event_time) FROM events;

-- 错误示例4: 聚合函数使用错误
-- 错误: SELECT user_id, COUNT(*) FROM events; (缺少GROUP BY)
-- 正确:
SELECT user_id, COUNT(*) FROM events GROUP BY user_id;

-- 使用VALIDATE语句进行预检查
VALIDATE SELECT * FROM non_existent_table;  -- 表不存在错误
VALIDATE SELECT invalid_function();         -- 函数不存在错误

2.2 复杂查询调试技巧

多层嵌套查询的问题定位。

-- 分步调试复杂查询
-- 原始复杂查询（难以调试）
SELECT 
    user_id,
    AVG(amount) AS avg_amount,
    COUNT(DISTINCT product_id) AS unique_products
FROM (
    SELECT 
        user_id,
        product_id,
        amount,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) as rn
    FROM transactions
    WHERE amount > 0 AND event_time > '2023-01-01'
) 
WHERE rn <= 10
GROUP BY user_id;

-- 分步调试版本
-- 步骤1: 检查源数据
SELECT COUNT(*) AS total_records FROM transactions 
WHERE amount > 0 AND event_time > '2023-01-01';

-- 步骤2: 验证窗口函数
SELECT 
    user_id,
    product_id,
    amount,
    event_time,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) as rn
FROM transactions
WHERE amount > 0 AND event_time > '2023-01-01'
LIMIT 100;

-- 步骤3: 验证过滤条件
SELECT COUNT(*) AS filtered_records
FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) as rn
    FROM transactions
    WHERE amount > 0 AND event_time > '2023-01-01'
) 
WHERE rn <= 10;

-- 步骤4: 最终聚合验证
SELECT 
    user_id,
    AVG(amount) AS avg_amount,
    COUNT(DISTINCT product_id) AS unique_products
FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) as rn
    FROM transactions
    WHERE amount > 0 AND event_time > '2023-01-01'
) 
WHERE rn <= 10
GROUP BY user_id
LIMIT 10;

3. 执行计划分析与优化

3.1 执行计划解读

理解查询执行路径。

-- 生成执行计划
EXPLAIN PLAN FOR
SELECT 
    user_id,
    COUNT(*) AS event_count,
    AVG(processing_time) AS avg_time
FROM user_events
WHERE event_time > CURRENT_TIMESTAMP - INTERVAL '1' HOUR
GROUP BY user_id;

-- 执行计划输出示例:
/*
== 优化后的物理计划 ==
Calc(select=[user_id, event_count, avg_time])
+- HashAggregate(isMerge=[true], groupBy=[user_id], select=[user_id, COUNT(count$0) AS event_count, AVG(sum$1, count$2) AS avg_time])
   +- Exchange(distribution=[hash[user_id]])
      +- LocalHashAggregate(groupBy=[user_id], select=[user_id, COUNT(*) AS count$0, AVG(processing_time) AS sum$1, COUNT(processing_time) AS count$2])
         +- Calc(select=[user_id, processing_time], where=[>(event_time, 2023-01-01 10:00:00)])
            +- TableSourceScan(table=[[default_catalog, default_database, user_events]], fields=[user_id, event_time, processing_time])
*/

-- 关键指标解读:
-- Exchange: 数据重分区操作（可能成为瓶颈）
-- HashAggregate: 哈希聚合操作（内存消耗点）
-- TableSourceScan: 表扫描操作（IO瓶颈点）

3.2 执行计划优化技巧

基于执行计划的性能调优。

-- 问题查询: 数据倾斜导致性能差
EXPLAIN PLAN FOR
SELECT 
    country,
    COUNT(DISTINCT user_id) AS unique_users
FROM user_sessions
GROUP BY country;

-- 优化方案1: 两阶段聚合解决数据倾斜
EXPLAIN PLAN FOR
SELECT 
    country,
    COUNT(*) AS unique_users
FROM (
    SELECT 
        country,
        user_id
    FROM user_sessions
    GROUP BY country, user_id  -- 先按复合键分组
) 
GROUP BY country;

-- 优化方案2: 使用Skew提示
SELECT /*+ SKEW('country') */
    country,
    COUNT(DISTINCT user_id) AS unique_users
FROM user_sessions
GROUP BY country;

-- 优化方案3: 调整并行度
SET 'table.exec.resource.default-parallelism' = '32';
EXPLAIN PLAN FOR
SELECT 
    country,
    COUNT(DISTINCT user_id) AS unique_users
FROM user_sessions
GROUP BY country;

4. 运行时故障排查

4.1 资源不足问题诊断

内存、CPU、网络资源瓶颈分析。

-- 内存溢出诊断查询
-- 监控状态大小
SELECT 
    TUMBLE_START(proc_time, INTERVAL '5' MINUTE) AS window_start,
    operator_id,
    state_name,
    SUM(state_size_bytes) AS total_state_size,
    MAX(state_size_bytes) AS max_state_size,
    COUNT(*) AS state_entries
FROM operator_state_metrics
GROUP BY TUMBLE(proc_time, INTERVAL '5' MINUTE), operator_id, state_name
HAVING SUM(state_size_bytes) > 1073741824;  -- 超过1GB告警

-- CPU使用率监控
CREATE TABLE cpu_monitoring AS
SELECT
    TUMBLE_START(metric_time, INTERVAL '1' MINUTE) AS window_start,
    taskmanager_id,
    AVG(cpu_usage) AS avg_cpu_usage,
    MAX(cpu_usage) AS max_cpu_usage
FROM taskmanager_metrics
GROUP BY TUMBLE(metric_time, INTERVAL '1' MINUTE), taskmanager_id
HAVING AVG(cpu_usage) > 80.0;  -- CPU使用率超过80%告警

-- 网络反压检测
CREATE TABLE backpressure_detection AS
SELECT
    CURRENT_TIMESTAMP AS detect_time,
    source_task_id,
    target_task_id,
    buffer_usage_ratio,
    CASE 
        WHEN buffer_usage_ratio > 0.8 THEN 'HIGH'
        WHEN buffer_usage_ratio > 0.5 THEN 'MEDIUM' 
        ELSE 'LOW'
    END AS backpressure_level
FROM network_metrics
WHERE buffer_usage_ratio > 0.5;

4.2 状态后端故障处理

RocksDB、内存状态问题排查。

-- RocksDB状态问题诊断
-- 1. 检查点失败分析
CREATE TABLE checkpoint_failures AS
SELECT
    checkpoint_id,
    failure_time,
    failure_cause,
    state_size_bytes,
    duration_ms,
    -- 失败原因分类
    CASE 
        WHEN failure_cause LIKE '%timeout%' THEN 'CHECKPOINT_TIMEOUT'
        WHEN failure_cause LIKE '%memory%' THEN 'MEMORY_OVERFLOW'
        WHEN failure_cause LIKE '%disk%' THEN 'DISK_FULL'
        ELSE 'OTHER'
    END AS failure_type
FROM checkpoint_history
WHERE status = 'FAILED';

-- 2. RocksDB压缩问题监控
CREATE TABLE rocksdb_compaction_issues AS
SELECT
    monitor_time,
    sst_file_count,
    pending_compaction_bytes,
    compression_ratio,
    -- 压缩健康度评估
    CASE 
        WHEN sst_file_count > 1000 THEN 'TOO_MANY_SST_FILES'
        WHEN pending_compaction_bytes > 1073741824 THEN 'COMPACTION_BACKLOG'  -- 1GB
        WHEN compression_ratio < 0.5 THEN 'POOR_COMPRESSION'
        ELSE 'HEALTHY'
    END AS compaction_health
FROM rocksdb_metrics;

-- 3. 状态访问延迟监控
CREATE TABLE state_access_latency AS
SELECT
    TUMBLE_START(access_time, INTERVAL '1' MINUTE) AS window_start,
    operator_id,
    state_name,
    PERCENTILE(access_latency_ms, 0.95) AS p95_latency,
    PERCENTILE(access_latency_ms, 0.99) AS p99_latency,
    AVG(access_latency_ms) AS avg_latency
FROM state_access_logs
GROUP BY TUMBLE(access_time, INTERVAL '1' MINUTE), operator_id, state_name
HAVING PERCENTILE(access_latency_ms, 0.95) > 1000;  -- P95延迟超过1秒告警

5. 数据质量与一致性排查

5.1 数据完整性验证

数据丢失、重复、乱序问题检测。

-- 数据完整性监控框架
CREATE TABLE data_quality_metrics (
    check_time TIMESTAMP(3),
    check_type STRING,
    source_count BIGINT,
    target_count BIGINT, 
    discrepancy BIGINT,
    completeness_ratio DOUBLE,
    is_healthy BOOLEAN
) WITH ('connector' = 'jdbc');

-- 1. 端到端数据完整性检查
INSERT INTO data_quality_metrics
SELECT
    CURRENT_TIMESTAMP,
    'end_to_end_completeness' AS check_type,
    source.record_count AS source_count,
    sink.record_count AS target_count,
    source.record_count - sink.record_count AS discrepancy,
    sink.record_count * 100.0 / source.record_count AS completeness_ratio,
    sink.record_count * 100.0 / source.record_count > 99.9 AS is_healthy
FROM (
    SELECT COUNT(*) AS record_count FROM source_table
    WHERE event_time > CURRENT_TIMESTAMP - INTERVAL '1' HOUR
) source
CROSS JOIN (
    SELECT COUNT(*) AS record_count FROM sink_table  
    WHERE process_time > CURRENT_TIMESTAMP - INTERVAL '1' HOUR
) sink;

-- 2. 重复数据检测
CREATE TABLE duplicate_detection AS
SELECT
    event_id,
    COUNT(*) AS occurrence_count,
    MIN(event_time) AS first_occurrence,
    MAX(event_time) AS last_occurrence
FROM events
GROUP BY event_id
HAVING COUNT(*) > 1;

-- 3. 乱序数据检测
-- 监控乱序事件的统计
CREATE TABLE out_of_order_metrics AS
SELECT
    window_start,
    window_end,
    -- 乱序事件统计
    COUNT_IF(is_out_of_order) AS out_of_order_count,
    COUNT(*) AS total_events,
    ROUND(COUNT_IF(is_out_of_order) * 100.0 / COUNT(*), 2) AS out_of_order_rate,
    -- 乱序程度统计
    MAX(out_of_order_seconds) AS max_out_of_order,
    AVG(out_of_order_seconds) AS avg_out_of_order,
    -- 按乱序级别统计
    COUNT_IF(out_of_order_seconds > 300) AS severe_out_of_order,
    COUNT_IF(out_of_order_seconds > 60) AS moderate_out_of_order,
    COUNT_IF(out_of_order_seconds > 0) AS slight_out_of_order
FROM (
    SELECT
        TUMBLE_START(process_time, INTERVAL '5' MINUTE) AS window_start,
        TUMBLE_END(process_time, INTERVAL '5' MINUTE) AS window_end,
        event_time,
        LAG(event_time) OVER w AS prev_event_time,
        CASE 
            WHEN LAG(event_time) OVER w IS NOT NULL 
            AND event_time < LAG(event_time) OVER w
            THEN DATEDIFF(SECOND, event_time, LAG(event_time) OVER w)
            ELSE 0
        END AS out_of_order_seconds,
        CASE 
            WHEN LAG(event_time) OVER w IS NOT NULL 
            AND event_time < LAG(event_time) OVER w
            THEN TRUE
            ELSE FALSE
        END AS is_out_of_order
    FROM events
    WINDOW w AS (
        PARTITION BY source_system, device_id
        ORDER BY process_time
        ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
    )
)
GROUP BY window_start, window_end;

5.2 水位线与时间语义问题

事件时间处理相关故障排查。

-- 水位线健康度监控
CREATE TABLE watermark_health AS
SELECT
    TUMBLE_START(proc_time, INTERVAL '1' MINUTE) AS window_start,
    source_id,
    -- 当前水位线
    CURRENT_WATERMARK(event_time) AS current_watermark,
    -- 最大事件时间
    MAX(event_time) AS max_event_time,
    -- 水位线延迟
    MAX(event_time) - CURRENT_WATERMARK(event_time) AS watermark_lag,
    -- 延迟事件数量
    SUM(CASE WHEN event_time < CURRENT_WATERMARK(event_time) THEN 1 ELSE 0 END) AS late_events,
    -- 健康度评估
    CASE 
        WHEN watermark_lag > INTERVAL '10' MINUTE THEN 'STALLED'
        WHEN watermark_lag > INTERVAL '1' MINUTE THEN 'LAGGING'
        ELSE 'HEALTHY'
    END AS health_status
FROM event_stream
GROUP BY TUMBLE(proc_time, INTERVAL '1' MINUTE), source_id;

-- 时间窗口完整性检查
CREATE TABLE window_completeness AS
SELECT
    window_start,
    window_end,
    expected_count,
    actual_count,
    actual_count * 100.0 / expected_count AS completeness_ratio,
    -- 基于历史数据预测期望数量
    LAG(actual_count, 1) OVER (ORDER BY window_start) AS previous_count
FROM (
    SELECT
        window_start,
        window_end, 
        -- 简单预测：使用前一个窗口的数量
        LAG(COUNT(*), 1) OVER (ORDER BY window_start) AS expected_count,
        COUNT(*) AS actual_count
    FROM (
        SELECT 
            window_start,
            window_end
        FROM TUMBLE(events, DESCRIPTOR(event_time), INTERVAL '10' MINUTES)
        GROUP BY window_start,
        		 window_end
    )
);

6. 连接器与外部系统故障

6.1 连接器故障诊断

Kafka、JDBC、文件系统等连接器问题。

-- Kafka连接器故障诊断
CREATE TABLE kafka_connector_issues AS
SELECT
    monitor_time,
    connector_id,
    -- 消费延迟监控
    current_offset,
    end_offset,
    end_offset - current_offset AS lag_records,
    (end_offset - current_offset) * 100.0 / end_offset AS lag_ratio,
    -- 连接状态
    is_connected,
    last_poll_time,
    CASE 
        WHEN NOT is_connected THEN 'DISCONNECTED'
        WHEN end_offset - current_offset > 10000 THEN 'HIGH_LAG'
        WHEN CURRENT_TIMESTAMP - last_poll_time > INTERVAL '1' MINUTE THEN 'STALLED'
        ELSE 'HEALTHY'
    END AS health_status
FROM kafka_consumer_metrics;

-- JDBC连接器故障诊断  
CREATE TABLE jdbc_connector_issues AS
SELECT
    check_time,
    connection_url,
    -- 连接池状态
    active_connections,
    idle_connections,
    total_connections,
    -- 执行性能
    avg_query_time_ms,
    error_count,
    timeout_count,
    -- 健康度评估
    CASE 
        WHEN error_count > 10 THEN 'HIGH_ERROR_RATE'
        WHEN avg_query_time_ms > 5000 THEN 'SLOW_QUERIES'
        WHEN active_connections = total_connections THEN 'CONNECTION_EXHAUSTED'
        ELSE 'HEALTHY'
    END AS health_status
FROM jdbc_connection_metrics;

-- 文件系统连接器监控
CREATE TABLE filesystem_connector_issues AS
SELECT
    check_time,
    base_path,
    -- 文件系统状态
    total_space_bytes,
    used_space_bytes,
    free_space_bytes,
    free_space_bytes * 100.0 / total_space_bytes AS free_ratio,
    -- 文件操作状态
    pending_files,
    in_progress_files,
    completed_files,
    failed_files,
    -- 空间不足预警
    CASE 
        WHEN free_ratio < 5.0 THEN 'CRITICAL'
        WHEN free_ratio < 10.0 THEN 'WARNING'
        ELSE 'HEALTHY'
    END AS disk_status
FROM filesystem_metrics;

6.2 序列化格式问题排查

JSON、Avro、Protobuf等格式错误。

-- 数据格式错误检测
CREATE TABLE format_error_detection AS
SELECT
    error_time,
    source_topic,
    error_type,
    error_message,
    malformed_data,
    -- 错误分类
    CASE 
        WHEN error_message LIKE '%JSON%' THEN 'JSON_PARSE_ERROR'
        WHEN error_message LIKE '%Avro%' THEN 'AVRO_SCHEMA_ERROR' 
        WHEN error_message LIKE '%Protobuf%' THEN 'PROTOBUF_DECODE_ERROR'
        ELSE 'OTHER_FORMAT_ERROR'
    END AS format_error_category
FROM deserialization_errors;

-- 模式演进兼容性检查
CREATE TABLE schema_compatibility AS
SELECT
    check_time,
    topic_name,
    current_schema_version,
    producer_schema_version,
    consumer_schema_version,
    -- 兼容性检查
    CASE 
        WHEN producer_schema_version > consumer_schema_version THEN 'PRODUCER_AHEAD'
        WHEN consumer_schema_version > producer_schema_version THEN 'CONSUMER_AHEAD' 
        WHEN producer_schema_version = consumer_schema_version THEN 'IN_SYNC'
        ELSE 'UNKNOWN'
    END AS compatibility_status
FROM schema_registry_metrics;

7. 性能优化实战案例

7.1 慢查询优化案例

真实场景性能问题解决。

-- 案例1: 数据倾斜导致的慢查询
-- 原始问题查询（某个城市数据量极大）
SELECT 
    city,
    COUNT(DISTINCT user_id) AS unique_users
FROM user_visits
WHERE visit_date = CURRENT_DATE
GROUP BY city;

-- 优化方案: 两阶段聚合 + 倾斜处理
WITH stage1 AS (
    -- 第一阶段: 局部聚合+随机分桶
    SELECT 
        city,
        user_id,
        -- 为数据倾斜的城市添加随机分桶
        CASE 
            WHEN city IN ('北京', '上海', '广州', '深圳') THEN 
                city || '_' || CAST(MOD(ABS(HASH(user_id)), 10) AS STRING)
            ELSE city
        END AS distributed_city
    FROM user_visits
    WHERE visit_date = CURRENT_DATE
    GROUP BY city, user_id, distributed_city
),
stage2 AS (
    -- 第二阶段: 全局聚合
    SELECT 
        -- 还原原始城市名称
        CASE 
            WHEN POSITION('_' IN distributed_city) > 0 THEN
                SUBSTRING(distributed_city, 1, POSITION('_' IN distributed_city) - 1)
            ELSE distributed_city
        END AS city,
        COUNT(*) AS unique_users
    FROM stage1
    GROUP BY distributed_city
)
SELECT city, SUM(unique_users) AS unique_users
FROM stage2
GROUP BY city;

-- 案例2: 大状态导致的Checkpoint超时
-- 优化前: 状态过大导致Checkpoint失败
-- 优化方案: 状态TTL + 增量Checkpoint

-- 启用状态TTL
CREATE TABLE user_sessions (
    user_id BIGINT,
    session_data STRING,
    last_activity TIMESTAMP(3),
    PRIMARY KEY (user_id) NOT ENFORCED
) /*+ STATE_TTL('7 days') */;  -- 7天自动清理

-- 优化RocksDB配置
SET 'state.backend.rocksdb.block.cache-size' = '256m';
SET 'state.backend.rocksdb.writebuffer.size' = '64m';
SET 'state.backend.rocksdb.compaction.style' = 'level';

7.2 资源优化案例

内存、CPU、网络资源调优。

-- 内存优化案例: 频繁Full GC问题
-- 监控内存使用
CREATE TABLE memory_usage_monitoring AS
SELECT
    TUMBLE_START(metric_time, INTERVAL '1' MINUTE) AS window_start,
    taskmanager_id,
    -- 内存使用情况
    heap_used_mb,
    heap_max_mb, 
    non_heap_used_mb,
    direct_memory_used_mb,
    -- GC情况
    gc_count,
    gc_time_ms,
    -- 内存压力评估
    CASE 
        WHEN heap_used_mb * 100.0 / heap_max_mb > 90 THEN 'CRITICAL'
        WHEN heap_used_mb * 100.0 / heap_max_mb > 80 THEN 'HIGH'
        WHEN gc_time_ms > 5000 THEN 'HIGH_GC'  -- 5秒GC时间
        ELSE 'HEALTHY'
    END AS memory_pressure
FROM jvm_metrics;

-- 优化方案: 调整内存配置
SET 'taskmanager.memory.task.heap.size' = '2g';        -- 增加堆内存
SET 'taskmanager.memory.managed.size' = '4g';           -- 增加托管内存
SET 'taskmanager.memory.jvm-metaspace.size' = '512m';   -- 增加元空间

-- CPU优化案例: 热点算子识别
CREATE TABLE cpu_hotspot_detection AS
SELECT
    TUMBLE_START(metric_time, INTERVAL '1' MINUTE) AS window_start,
    operator_id,
    task_id,
    cpu_time_ms,
    records_processed,
    cpu_time_ms * 1000.0 / records_processed AS us_per_record,  -- 每记录耗时
    -- 热点识别
    CASE 
        WHEN cpu_time_ms * 1000.0 / records_processed > 1000 THEN 'VERY_HOT'  -- 1ms/record
        WHEN cpu_time_ms * 1000.0 / records_processed > 100 THEN 'HOT'        -- 100us/record
        ELSE 'NORMAL'
    END AS hotspot_level
FROM operator_metrics
WHERE records_processed > 0;

8. 自动化诊断与修复

8.1 智能诊断系统

基于规则的自动问题检测。

-- 自动化诊断规则引擎
CREATE TABLE diagnostic_rules (
    rule_id STRING,
    rule_name STRING,
    rule_condition STRING,
    severity STRING,  -- CRITICAL, HIGH, MEDIUM, LOW
    suggested_action STRING
) WITH ('connector' = 'jdbc');

INSERT INTO diagnostic_rules VALUES
('RULE_001', '高水位线延迟', 'watermark_lag > INTERVAL ''5'' MINUTE', 'HIGH', 
 '检查源端数据延迟或调整水位线间隔'),
('RULE_002', 'Checkpoint超时', 'checkpoint_duration > checkpoint_timeout', 'CRITICAL',
 '增加Checkpoint超时时间或优化状态后端'),
('RULE_003', '数据倾斜', 'max_records_per_task > 2 * avg_records_per_task', 'MEDIUM',
 '使用两阶段聚合或添加随机前缀');

-- 自动诊断执行
CREATE TABLE automated_diagnostics AS
SELECT
    CURRENT_TIMESTAMP AS diagnosis_time,
    rule.rule_id,
    rule.rule_name,
    rule.severity,
    rule.suggested_action,
    metrics.*
FROM diagnostic_rules rule
JOIN system_metrics metrics ON eval(rule.rule_condition)  -- 规则条件评估
WHERE rule.severity IN ('CRITICAL', 'HIGH');

8.2 自愈修复机制

自动化问题修复策略。

-- 自愈动作定义
CREATE TABLE healing_actions (
    action_id STRING,
    trigger_condition STRING,
    action_type STRING,  -- RESTART, RECONFIGURE, SCALE
    action_parameters STRING,
    cooldown_minutes INT
) WITH ('connector' = 'jdbc');

-- 自愈执行监控
CREATE TABLE healing_execution_log AS
SELECT
    CURRENT_TIMESTAMP AS execution_time,
    action.action_id,
    action.action_type,
    diagnosis.issue_details,
    -- 执行结果跟踪
    CASE 
        WHEN action_success = true THEN 'SUCCESS'
        ELSE 'FAILED'
    END AS execution_result,
    action_duration_ms
FROM healing_actions action
JOIN current_issues diagnosis ON eval(action.trigger_condition)
WHERE last_execution_time IS NULL 
   OR CURRENT_TIMESTAMP - last_execution_time > INTERVAL '5' MINUTE;  -- 冷却期检查

9. 总结

SQL作业故障排查需要系统化的方法论和工具链支持。从语法错误到运行时故障，从数据质量到性能瓶颈，每个层面都需要特定的诊断技术。关键成功要素包括：
分层诊断：从SQL语法到基础设施的逐层排查
执行计划分析：理解查询执行路径，识别性能热点
监控体系：建立完整的指标收集和告警机制
数据质量：端到端的数据完整性验证
自动化：规则引擎驱动的智能诊断和自愈
通过科学的排查流程和自动化工具，可以大幅提升故障解决效率，保障流处理作业的稳定运行。