-- 實(shí)時(shí)交易流  
CREATE TABLE transaction_stream (  
    transaction_id BIGINT,  
    user_id BIGINT,  
    amount DECIMAL(15,2),  
    merchant_id BIGINT,  
    transaction_type STRING,  
    transaction_time TIMESTAMP(3),  
    ip_address STRING,  
    device_id STRING,  
    location STRING,  
    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '10' SECOND  
) WITH (  
    'connector' = 'kafka',  
    'topic' = 'financial-transactions',  
    'properties.bootstrap.servers' = 'localhost:9092',  
    'format' = 'json',  
    'scan.startup.mode' = 'latest-offset'  
);

-- 用戶風(fēng)險(xiǎn)檔案表  
CREATE TABLE user_risk_profile (  
    user_id BIGINT,  
    risk_score INT,  
    risk_level STRING,  
    credit_rating STRING,  
    account_age_days INT,  
    avg_daily_amount DECIMAL(15,2),  
    max_single_amount DECIMAL(15,2),  
    suspicious_activity_count INT,  
    last_update_time TIMESTAMP(3),  
    PRIMARY KEY (user_id) NOT ENFORCED  
) WITH (  
    'connector' = 'jdbc',  
    'url' = 'jdbc:mysql://localhost:3306/risk_db',  
    'table-name' = 'user_risk_profiles',  
    'lookup.cache.max-rows' = '100000',  
    'lookup.cache.ttl' = '30min'  
);  


-- 商戶風(fēng)險(xiǎn)檔案表  
CREATE TABLE merchant_risk_profile (  
    merchant_id BIGINT,  
    merchant_category STRING,  
    risk_level STRING,  
    fraud_rate DECIMAL(5,4),  
    avg_transaction_amount DECIMAL(15,2),  
    business_hours_start TIME,  
    business_hours_end TIME,  
    PRIMARY KEY (merchant_id) NOT ENFORCED  
) WITH (  
    'connector' = 'jdbc',  
    'url' = 'jdbc:mysql://localhost:3306/risk_db',  
    'table-name' = 'merchant_risk_profiles',  
    'lookup.cache.max-rows' = '50000',  
    'lookup.cache.ttl' = '1h'  
);

-- 實(shí)時(shí)風(fēng)險(xiǎn)評(píng)分計(jì)算  
CREATE VIEW real_time_risk_scoring AS  
SELECT /*+ BROADCAST(user_risk_profile) */  
    t.transaction_id,  
    t.user_id,  
    t.amount,  
    t.merchant_id,  
    t.transaction_time,  
    t.ip_address,  
    t.device_id,  
    u.risk_level as user_risk_level,  
    u.risk_score as base_risk_score,  
    m.risk_level as merchant_risk_level,  
    m.fraud_rate as merchant_fraud_rate,  


    -- 金額異常評(píng)分  
    CASE   
        WHEN t.amount > u.max_single_amount * 2 THEN 50  
        WHEN t.amount > u.avg_daily_amount * 10 THEN 30  
        WHEN t.amount > u.avg_daily_amount * 5 THEN 20  
        ELSE 0  
    END as amount_anomaly_score,  


    -- 時(shí)間異常評(píng)分  
    CASE   
        WHEN EXTRACT(HOUR FROM t.transaction_time) BETWEEN 2 AND 5 THEN 15  
        WHEN EXTRACT(HOUR FROM t.transaction_time) NOT BETWEEN   
             EXTRACT(HOUR FROM m.business_hours_start) AND   
             EXTRACT(HOUR FROM m.business_hours_end) THEN 10  
        ELSE 0  
    END as time_anomaly_score,  


    -- 綜合風(fēng)險(xiǎn)評(píng)分  
    u.risk_score +   
    CASE   
        WHEN t.amount > u.max_single_amount * 2 THEN 50  
        WHEN t.amount > u.avg_daily_amount * 10 THEN 30  
        WHEN t.amount > u.avg_daily_amount * 5 THEN 20  
        ELSE 0  
    END +  
    CASE   
        WHEN EXTRACT(HOUR FROM t.transaction_time) BETWEEN 2 AND 5 THEN 15  
        WHEN EXTRACT(HOUR FROM t.transaction_time) NOT BETWEEN   
             EXTRACT(HOUR FROM m.business_hours_start) AND   
             EXTRACT(HOUR FROM m.business_hours_end) THEN 10  
        ELSE 0  
    END as total_risk_score  


FROM transaction_stream t  
LEFT JOIN user_risk_profile FOR SYSTEM_TIME AS OF t.transaction_time AS u  
ON t.user_id = u.user_id  
LEFT JOIN merchant_risk_profile FOR SYSTEM_TIME AS OF t.transaction_time AS m  
ON t.merchant_id = m.merchant_id;

-- 用戶行為模式分析  
CREATE VIEW user_behavior_analysis AS  
SELECT   
    user_id,  
    transaction_time,  
    -- 近1小時(shí)交易頻次  
    COUNT(*) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW  
    ) as txn_count_1h,  


    -- 近1小時(shí)交易總額  
    SUM(amount) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW  
    ) as txn_amount_1h,  


    -- 近24小時(shí)不同IP數(shù)量  
    COUNT(DISTINCT ip_address) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '24' HOUR PRECEDING AND CURRENT ROW  
    ) as distinct_ip_24h,  


    -- 近24小時(shí)不同設(shè)備數(shù)量  
    COUNT(DISTINCT device_id) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '24' HOUR PRECEDING AND CURRENT ROW  
    ) as distinct_device_24h,  


    -- 連續(xù)小額交易檢測(cè)  
    CASE   
        WHEN amount < 100 AND   
             LAG(amount, 1) OVER (PARTITION BY user_id ORDER BY transaction_time) < 100 AND  
             LAG(amount, 2) OVER (PARTITION BY user_id ORDER BY transaction_time) < 100  
        THEN 1 ELSE 0  
    END as small_amount_pattern,  


    transaction_id,  
    amount,  
    merchant_id  
FROM transaction_stream;

-- 歷史交易模式對(duì)比  
CREATE TABLE transaction_history_summary (  
    user_id BIGINT,  
    date_key DATE,  
    hour_key INT,  
    total_amount DECIMAL(15,2),  
    transaction_count INT,  
    avg_amount DECIMAL(15,2),  
    max_amount DECIMAL(15,2),  
    distinct_merchants INT,  
    most_frequent_merchant BIGINT  
) WITH (  
    'connector' = 'filesystem',  
    'path' = 'hdfs://namenode:9000/data/transaction_history',  
    'format' = 'parquet'  
);  


-- 歷史模式異常檢測(cè)  
CREATE VIEW historical_pattern_analysis AS  
SELECT /*+ SHUFFLE_MERGE(user_behavior_analysis, transaction_history_summary) */  
    b.transaction_id,  
    b.user_id,  
    b.amount,  
    b.transaction_time,  
    b.txn_count_1h,  
    b.txn_amount_1h,  
    h.avg_amount as historical_avg,  
    h.max_amount as historical_max,  
    h.transaction_count as historical_count,  


    -- 交易頻次異常  
    CASE   
        WHEN b.txn_count_1h > h.transaction_count * 3 THEN 'HIGH_FREQUENCY_ANOMALY'  
        WHEN b.txn_count_1h > h.transaction_count * 2 THEN 'MEDIUM_FREQUENCY_ANOMALY'  
        ELSE 'NORMAL_FREQUENCY'  
    END as frequency_anomaly,  


    -- 交易金額異常  
    CASE   
        WHEN b.amount > h.max_amount * 2 THEN 'EXTREME_AMOUNT_ANOMALY'  
        WHEN b.amount > h.avg_amount * 10 THEN 'HIGH_AMOUNT_ANOMALY'  
        WHEN ABS(b.amount - h.avg_amount) / h.avg_amount > 5 THEN 'AMOUNT_DEVIATION_ANOMALY'  
        ELSE 'NORMAL_AMOUNT'  
    END as amount_anomaly,  


    -- 設(shè)備/IP異常  
    CASE   
        WHEN b.distinct_ip_24h > 5 THEN 'MULTIPLE_IP_RISK'  
        WHEN b.distinct_device_24h > 3 THEN 'MULTIPLE_DEVICE_RISK'  
        ELSE 'NORMAL_ACCESS'  
    END as access_anomaly  


FROM user_behavior_analysis b  
LEFT JOIN transaction_history_summary h   
ON b.user_id = h.user_id   
AND DATE(b.transaction_time) = h.date_key  
AND EXTRACT(HOUR FROM b.transaction_time) = h.hour_key;

-- 綜合風(fēng)控決策引擎  
CREATE VIEW comprehensive_risk_decision AS  
SELECT   
    r.transaction_id,  
    r.user_id,  
    r.amount,  
    r.merchant_id,  
    r.transaction_time,  
    r.total_risk_score,  
    r.user_risk_level,  
    r.merchant_risk_level,  
    h.frequency_anomaly,  
    h.amount_anomaly,  
    h.access_anomaly,  


    -- 黑名單檢查  
    CASE   
        WHEN r.user_id IN (SELECT user_id FROM blacklist_users) THEN 'BLACKLIST_USER'  
        WHEN r.merchant_id IN (SELECT merchant_id FROM blacklist_merchants) THEN 'BLACKLIST_MERCHANT'  
        WHEN r.ip_address IN (SELECT ip_address FROM blacklist_ips) THEN 'BLACKLIST_IP'  
        ELSE 'NOT_BLACKLISTED'  
    END as blacklist_status,  


    -- 地理位置風(fēng)險(xiǎn)  
    CASE   
        WHEN r.location IN ('高風(fēng)險(xiǎn)國(guó)家1', '高風(fēng)險(xiǎn)國(guó)家2') THEN 'HIGH_GEO_RISK'  
        WHEN r.location != LAG(r.location) OVER (  
            PARTITION BY r.user_id   
            ORDER BY r.transaction_time  
        ) THEN 'LOCATION_CHANGE_RISK'  
        ELSE 'NORMAL_GEO'  
    END as geo_risk,  


    -- 最終決策  
    CASE   
        -- 立即拒絕條件  
        WHEN r.total_risk_score > 100 OR  
             r.user_id IN (SELECT user_id FROM blacklist_users) OR  
             h.frequency_anomaly = 'HIGH_FREQUENCY_ANOMALY' AND h.amount_anomaly = 'EXTREME_AMOUNT_ANOMALY'  
        THEN 'REJECT'  


        -- 人工審核條件  
        WHEN r.total_risk_score > 70 OR  
             r.user_risk_level = 'HIGH' OR  
             h.amount_anomaly IN ('HIGH_AMOUNT_ANOMALY', 'EXTREME_AMOUNT_ANOMALY') OR  
             h.access_anomaly IN ('MULTIPLE_IP_RISK', 'MULTIPLE_DEVICE_RISK')  
        THEN 'MANUAL_REVIEW'  


        -- 延遲處理?xiàng)l件  
        WHEN r.total_risk_score > 50 OR  
             r.merchant_risk_level = 'HIGH' OR  
             h.frequency_anomaly = 'MEDIUM_FREQUENCY_ANOMALY'  
        THEN 'DELAYED_APPROVAL'  


        -- 正常通過(guò)  
        ELSE 'APPROVE'  
    END as final_decision,  


    -- 風(fēng)險(xiǎn)原因  
    CONCAT_WS('; ',  
        CASE WHEN r.total_risk_score > 70 THEN '高風(fēng)險(xiǎn)評(píng)分' END,  
        CASE WHEN h.amount_anomaly != 'NORMAL_AMOUNT' THEN '金額異常' END,  
        CASE WHEN h.frequency_anomaly != 'NORMAL_FREQUENCY' THEN '頻次異常' END,  
        CASE WHEN h.access_anomaly != 'NORMAL_ACCESS' THEN '訪問(wèn)異常' END  
    ) as risk_reasons  


FROM real_time_risk_scoring r  
JOIN historical_pattern_analysis h ON r.transaction_id = h.transaction_id;

-- 實(shí)時(shí)處理性能監(jiān)控  
CREATE VIEW system_performance_monitoring AS  
SELECT   
    TUMBLE_START(transaction_time, INTERVAL '1' MINUTE) as window_start,  
    COUNT(*) as total_transactions,  
    COUNT(CASE WHEN final_decision = 'REJECT' THEN 1 END) as rejected_count,  
    COUNT(CASE WHEN final_decision = 'MANUAL_REVIEW' THEN 1 END) as review_count,  
    COUNT(CASE WHEN final_decision = 'APPROVE' THEN 1 END) as approved_count,  


    -- 拒絕率  
    COUNT(CASE WHEN final_decision = 'REJECT' THEN 1 END) * 1.0 / COUNT(*) as rejection_rate,  


    -- 平均處理延遲（毫秒）  
    AVG(UNIX_TIMESTAMP() * 1000 - UNIX_TIMESTAMP(transaction_time) * 1000) as avg_processing_latency_ms,  


    -- 高風(fēng)險(xiǎn)交易占比  
    COUNT(CASE WHEN total_risk_score > 70 THEN 1 END) * 1.0 / COUNT(*) as high_risk_ratio  


FROM comprehensive_risk_decision  
GROUP BY TUMBLE(transaction_time, INTERVAL '1' MINUTE);

-- 異常告警觸發(fā)  
CREATE VIEW alert_triggers AS  
SELECT   
    window_start,  
    total_transactions,  
    rejection_rate,  
    avg_processing_latency_ms,  
    high_risk_ratio,  


    -- 告警級(jí)別判斷  
    CASE   
        WHEN rejection_rate > 0.3 OR avg_processing_latency_ms > 5000 THEN 'CRITICAL'  
        WHEN rejection_rate > 0.2 OR avg_processing_latency_ms > 3000 THEN 'HIGH'  
        WHEN rejection_rate > 0.1 OR avg_processing_latency_ms > 1000 THEN 'MEDIUM'  
        ELSE 'NORMAL'  
    END as alert_level,  


    -- 告警消息  
    CASE   
        WHEN rejection_rate > 0.3 THEN CONCAT('拒絕率過(guò)高: ', CAST(rejection_rate * 100 AS STRING), '%')  
        WHEN avg_processing_latency_ms > 5000 THEN CONCAT('處理延遲過(guò)高: ', CAST(avg_processing_latency_ms AS STRING), 'ms')  
        WHEN high_risk_ratio > 0.5 THEN CONCAT('高風(fēng)險(xiǎn)交易占比過(guò)高: ', CAST(high_risk_ratio * 100 AS STRING), '%')  
        ELSE 'Normal'  
    END as alert_message  


FROM system_performance_monitoring  
WHERE rejection_rate > 0.1 OR avg_processing_latency_ms > 1000 OR high_risk_ratio > 0.3;

-- 實(shí)時(shí)特征提取  
CREATE VIEW ml_feature_extraction AS  
SELECT   
    transaction_id,  
    user_id,  
    amount,  
    merchant_id,  
    transaction_time,  


    -- 用戶歷史特征  
    AVG(amount) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '30' DAY PRECEDING AND CURRENT ROW  
    ) as user_avg_amount_30d,  


    STDDEV(amount) OVER (  
        PARTITION BY user_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '30' DAY PRECEDING AND CURRENT ROW  
    ) as user_amount_stddev_30d,  


    -- 商戶特征  
    AVG(amount) OVER (  
        PARTITION BY merchant_id   
        ORDER BY transaction_time   
        RANGE BETWEEN INTERVAL '7' DAY PRECEDING AND CURRENT ROW  
    ) as merchant_avg_amount_7d,  


    -- 時(shí)間特征  
    EXTRACT(HOUR FROM transaction_time) as hour_of_day,  
    EXTRACT(DOW FROM transaction_time) as day_of_week,  


    -- 交易間隔特征  
    UNIX_TIMESTAMP(transaction_time) -   
    UNIX_TIMESTAMP(LAG(transaction_time) OVER (PARTITION BY user_id ORDER BY transaction_time)) as time_since_last_txn,  


    -- 金額比率特征  
    amount / NULLIF(LAG(amount) OVER (PARTITION BY user_id ORDER BY transaction_time), 0) as amount_ratio_to_prev  


FROM transaction_stream;

// 機(jī)器學(xué)習(xí)模型預(yù)測(cè)函數(shù)  
public class MLRiskPredictionFunction extends RichMapFunction<Transaction, TransactionWithMLScore> {  
    private transient MLModel riskModel;  
    private transient FeatureExtractor featureExtractor;  


    @Override  
    public void open(Configuration parameters) throws Exception {  
        // 加載預(yù)訓(xùn)練的風(fēng)險(xiǎn)預(yù)測(cè)模型  
        this.riskModel = MLModelLoader.loadModel("risk-prediction-model-v2.pkl");  
        this.featureExtractor = new FeatureExtractor();  
    }  


    @Override  
    public TransactionWithMLScore map(Transaction transaction) throws Exception {  
        // 提取特征向量  
        double[] features = featureExtractor.extractFeatures(transaction);  


        // 模型預(yù)測(cè)  
        double riskProbability = riskModel.predict(features);  
        String riskCategory = categorizeRisk(riskProbability);  


        return new TransactionWithMLScore(  
            transaction,  
            riskProbability,  
            riskCategory,  
            System.currentTimeMillis()  
        );  
    }  


    private String categorizeRisk(double probability) {  
        if (probability > 0.8) return "HIGH_RISK";  
        if (probability > 0.6) return "MEDIUM_RISK";  
        if (probability > 0.3) return "LOW_RISK";  
        return "VERY_LOW_RISK";  
    }  
}

// 主要的Flink作業(yè)配置  
public class FinancialRiskControlJob {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  


        // 配置檢查點(diǎn)  
        env.enableCheckpointing(60000); // 1分鐘檢查點(diǎn)  
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);  
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);  


        // 配置狀態(tài)后端  
        env.setStateBackend(new HashMapStateBackend());  
        env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:9000/flink-checkpoints");  


        // 設(shè)置并行度  
        env.setParallelism(16);  


        // 創(chuàng)建表環(huán)境  
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);  


        // 注冊(cè)數(shù)據(jù)源表  
        registerSourceTables(tableEnv);  


        // 注冊(cè)維度表  
        registerDimensionTables(tableEnv);  


        // 執(zhí)行風(fēng)控邏輯  
        executeRiskControlLogic(tableEnv);  


        env.execute("Financial Risk Control Job");  
    }  


    private static void registerSourceTables(StreamTableEnvironment tableEnv) {  
        // 注冊(cè)交易流表  
        tableEnv.executeSql("""  
            CREATE TABLE transaction_stream (  
                transaction_id BIGINT,  
                user_id BIGINT,  
                amount DECIMAL(15,2),  
                merchant_id BIGINT,  
                transaction_type STRING,  
                transaction_time TIMESTAMP(3),  
                ip_address STRING,  
                device_id STRING,  
                location STRING,  
                WATERMARK FOR transaction_time AS transaction_time - INTERVAL '10' SECOND  
            ) WITH (  
                'connector' = 'kafka',  
                'topic' = 'financial-transactions',  
                'properties.bootstrap.servers' = 'localhost:9092',  
                'format' = 'json'  
            )  
        """);  
    }  
}

-- 使用BROADCAST優(yōu)化小表連接  
CREATE VIEW optimized_risk_scoring AS  
SELECT /*+ BROADCAST(user_risk_profile, merchant_risk_profile) */  
    t.transaction_id,  
    t.user_id,  
    t.amount,  
    u.risk_score,  
    m.fraud_rate,  
    -- 計(jì)算綜合風(fēng)險(xiǎn)評(píng)分  
    CASE   
        WHEN t.amount > u.max_single_amount * 2 THEN u.risk_score + 50  
        WHEN t.amount > u.avg_daily_amount * 5 THEN u.risk_score + 30  
        ELSE u.risk_score + 10  
    END as calculated_risk_score  
FROM transaction_stream t  
LEFT JOIN user_risk_profile FOR SYSTEM_TIME AS OF t.transaction_time AS u  
ON t.user_id = u.user_id  
LEFT JOIN merchant_risk_profile FOR SYSTEM_TIME AS OF t.transaction_time AS m  
ON t.merchant_id = m.merchant_id;

// 自定義狀態(tài)管理  
public class UserRiskStateFunction extends KeyedProcessFunction<Long, Transaction, RiskAssessment> {  


    // 用戶交易歷史狀態(tài)  
    private ValueState<UserTransactionHistory> userHistoryState;  


    // 滑動(dòng)窗口狀態(tài)  
    private MapState<Long, TransactionSummary> hourlyStatsState;  


    @Override  
    public void open(Configuration parameters) {  
        // 配置狀態(tài)描述符  
        ValueStateDescriptor<UserTransactionHistory> historyDescriptor =   
            new ValueStateDescriptor<>("user-history", UserTransactionHistory.class);  
        historyDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(30))  
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  
            .build());  


        userHistoryState = getRuntimeContext().getState(historyDescriptor);  


        MapStateDescriptor<Long, TransactionSummary> statsDescriptor =   
            new MapStateDescriptor<>("hourly-stats", Long.class, TransactionSummary.class);  
        hourlyStatsState = getRuntimeContext().getMapState(statsDescriptor);  
    }  


    @Override  
    public void processElement(Transaction transaction, Context ctx, Collector<RiskAssessment> out)   
            throws Exception {  


        // 獲取用戶歷史  
        UserTransactionHistory history = userHistoryState.value();  
        if (history == null) {  
            history = new UserTransactionHistory();  
        }  


        // 更新歷史記錄  
        history.addTransaction(transaction);  
        userHistoryState.update(history);  


        // 計(jì)算風(fēng)險(xiǎn)評(píng)分  
        RiskAssessment assessment = calculateRisk(transaction, history);  
        out.collect(assessment);  


        // 設(shè)置定時(shí)器清理過(guò)期數(shù)據(jù)  
        ctx.timerService().registerProcessingTimeTimer(  
            ctx.timerService().currentProcessingTime() + 3600000); // 1小時(shí)后清理  
    }  
}

-- 系統(tǒng)健康度監(jiān)控  
CREATE VIEW system_health_metrics AS  
SELECT   
    TUMBLE_START(transaction_time, INTERVAL '5' MINUTE) as window_start,  


    -- 吞吐量指標(biāo)  
    COUNT(*) as total_transactions,  
    COUNT(*) / 300.0 as tps, -- 每秒交易數(shù)  


    -- 延遲指標(biāo)  
    AVG(processing_latency_ms) as avg_latency,  
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY processing_latency_ms) as p95_latency,  
    PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY processing_latency_ms) as p99_latency,  


    -- 準(zhǔn)確性指標(biāo)  
    COUNT(CASE WHEN final_decision = 'REJECT' THEN 1 END) as rejected_count,  
    COUNT(CASE WHEN final_decision = 'MANUAL_REVIEW' THEN 1 END) as review_count,  


    -- 系統(tǒng)穩(wěn)定性  
    COUNT(CASE WHEN processing_error IS NOT NULL THEN 1 END) as error_count,  
    COUNT(CASE WHEN processing_error IS NOT NULL THEN 1 END) * 1.0 / COUNT(*) as error_rate  


FROM comprehensive_risk_decision  
GROUP BY TUMBLE(transaction_time, INTERVAL '5' MINUTE);

// 自動(dòng)化運(yùn)維管理器  
public class AutoOpsManager {  
    private final MetricsCollector metricsCollector;  
    private final AlertManager alertManager;  
    private final JobManager jobManager;  


    @Scheduled(fixedRate = 30000) // 每30秒檢查一次  
    public void performHealthCheck() {  
        SystemMetrics metrics = metricsCollector.collectSystemMetrics();  


        // 檢查吞吐量  
        if (metrics.getTps() < 100) {  
            alertManager.sendAlert(AlertLevel.HIGH, "TPS過(guò)低: " + metrics.getTps());  
        }  


        // 檢查延遲  
        if (metrics.getP99Latency() > 5000) {  
            alertManager.sendAlert(AlertLevel.CRITICAL, "P99延遲過(guò)高: " + metrics.getP99Latency() + "ms");  
        }  


        // 檢查錯(cuò)誤率  
        if (metrics.getErrorRate() > 0.01) {  
            alertManager.sendAlert(AlertLevel.HIGH, "錯(cuò)誤率過(guò)高: " + (metrics.getErrorRate() * 100) + "%");  


            // 自動(dòng)重啟作業(yè)  
            if (metrics.getErrorRate() > 0.05) {  
                jobManager.restartJob("financial-risk-control");  
            }  
        }  
    }  
}

apiVersion: apps/v1  
kind: Deployment  
metadata:  
  name: flink-risk-control  
spec:  
  replicas: 3  
  selector:  
    matchLabels:  
      app: flink-risk-control  
  template:  
    metadata:  
      labels:  
        app: flink-risk-control  
    spec:  
      containers:  
      - name: flink-taskmanager  
        image: flink:1.18  
        resources:  
          requests:  
            memory: "4Gi"  
            cpu: "2"  
          limits:  
            memory: "8Gi"  
            cpu: "4"  
        env:  
        - name: FLINK_PROPERTIES  
          value: |  
            jobmanager.rpc.address: flink-jobmanager  
            taskmanager.memory.process.size: 4096m  
            taskmanager.numberOfTaskSlots: 4  
            state.backend: hashmap  
            state.checkpoints.dir: hdfs://namenode:9000/flink-checkpoints  
            state.savepoints.dir: hdfs://namenode:9000/flink-savepoints  
            execution.checkpointing.interval: 60s  
            execution.checkpointing.mode: EXACTLY_ONCE  
            table.exec.source.idle-timeout: 30s  
        volumeMounts:  
        - name: flink-config  
          mountPath: /opt/flink/conf  
        - name: hadoop-config  
          mountPath: /etc/hadoop/conf  
      volumes:  
      - name: flink-config  
        configMap:  
          name: flink-config  
      - name: hadoop-config  
        configMap:  
          name: hadoop-config  
---  
apiVersion: v1  
kind: Service  
metadata:  
  name: flink-jobmanager  
spec:  
  type: ClusterIP  
  ports:  
  - name: rpc  
    port: 6123  
    targetPort: 6123  
  - name: blob-server  
    port: 6124  
    targetPort: 6124  
  - name: webui  
    port: 8081  
    targetPort: 8081  
  selector:  
    app: flink-jobmanager

# HorizontalPodAutoscaler配置  
apiVersion: autoscaling/v2  
kind: HorizontalPodAutoscaler  
metadata:  
  name: flink-risk-control-hpa  
spec:  
  scaleTargetRef:  
    apiVersion: apps/v1  
    kind: Deployment  
    name: flink-risk-control  
  minReplicas: 3  
  maxReplicas: 20  
  metrics:  
  - type: Resource  
    resource:  
      name: cpu  
      target:  
        type: Utilization  
        averageUtilization: 70  
  - type: Resource  
    resource:  
      name: memory  
      target:  
        type: Utilization  
        averageUtilization: 80  
  - type: Pods  
    pods:  
      metric:  
        name: flink_taskmanager_job_task_backPressuredTimeMsPerSecond  
      target:  
        type: AverageValue  
        averageValue: "1000"

// 敏感數(shù)據(jù)加密處理  
public class DataEncryptionFunction extends RichMapFunction<Transaction, EncryptedTransaction> {  
    private transient AESUtil encryptor;  
    private transient String encryptionKey;  


    @Override  
    public void open(Configuration parameters) throws Exception {  
        // 從安全配置中獲取加密密鑰  
        this.encryptionKey = parameters.getString("security.encryption.key");  
        this.encryptor = new AESUtil(encryptionKey);  
    }  


    @Override  
    public EncryptedTransaction map(Transaction transaction) throws Exception {  
        return EncryptedTransaction.builder()  
            .transactionId(transaction.getTransactionId())  
            .userId(encryptor.encrypt(String.valueOf(transaction.getUserId())))  
            .amount(transaction.getAmount())  
            .merchantId(transaction.getMerchantId())  
            .transactionTime(transaction.getTransactionTime())  
            // IP地址脫敏處理  
            .ipAddress(maskIpAddress(transaction.getIpAddress()))  
            // 設(shè)備ID哈希處理  
            .deviceId(hashDeviceId(transaction.getDeviceId()))  
            .location(transaction.getLocation())  
            .build();  
    }  


    private String maskIpAddress(String ipAddress) {  
        // 保留前兩段IP，后兩段用*替代  
        String[] parts = ipAddress.split("\\.");  
        if (parts.length == 4) {  
            return parts[0] + "." + parts[1] + ".*.*";  
        }  
        return "***.***.***.***.";  
    }  


    private String hashDeviceId(String deviceId) {  
        return DigestUtils.sha256Hex(deviceId + encryptionKey);  
    }  
}

-- 審計(jì)日志表  
CREATE TABLE audit_log (  
    log_id BIGINT,  
    transaction_id BIGINT,  
    user_id STRING, -- 加密后的用戶ID  
    operation_type STRING,  
    risk_decision STRING,  
    risk_score INT,  
    decision_reason STRING,  
    operator_id STRING,  
    operation_time TIMESTAMP(3),  
    ip_address STRING,  
    system_version STRING  
) WITH (  
    'connector' = 'elasticsearch',  
    'hosts' = 'http://elasticsearch:9200',  
    'index' = 'financial-audit-logs'  
);  


-- 審計(jì)日志寫(xiě)入  
INSERT INTO audit_log  
SELECT   
    UNIX_TIMESTAMP() * 1000 + ROW_NUMBER() OVER (ORDER BY transaction_time) as log_id,  
    transaction_id,  
    user_id,  
    'RISK_ASSESSMENT' as operation_type,  
    final_decision as risk_decision,  
    total_risk_score as risk_score,  
    risk_reasons as decision_reason,  
    'SYSTEM_AUTO' as operator_id,  
    transaction_time as operation_time,  
    ip_address,  
    '2.0.1' as system_version  
FROM comprehensive_risk_decision;

// 自動(dòng)故障切換管理器  
public class FailoverManager {  
    private final ClusterMonitor clusterMonitor;  
    private final JobManager jobManager;  
    private final ConfigurationManager configManager;  


    @Scheduled(fixedRate = 10000) // 每10秒檢查一次  
    public void performHealthCheck() {  
        ClusterHealth health = clusterMonitor.checkClusterHealth();  


        if (health.getJobManagerStatus() == Status.DOWN) {  
            log.warn("JobManager is down, initiating failover...");  
            initiateJobManagerFailover();  
        }  


        if (health.getTaskManagerCount() < health.getMinRequiredTaskManagers()) {  
            log.warn("Insufficient TaskManagers, scaling up...");  
            scaleUpTaskManagers();  
        }  


        if (health.getKafkaLag() > 100000) {  
            log.warn("High Kafka lag detected, checking for backpressure...");  
            handleBackpressure();  
        }  
    }  


    private void initiateJobManagerFailover() {  
        try {  
            // 1. 停止當(dāng)前作業(yè)  
            jobManager.cancelJob("financial-risk-control");  


            // 2. 切換到備用集群  
            configManager.switchToBackupCluster();  


            // 3. 從最新檢查點(diǎn)恢復(fù)作業(yè)  
            String latestCheckpoint = getLatestCheckpoint();  
            jobManager.restoreJobFromCheckpoint("financial-risk-control", latestCheckpoint);  


            log.info("Failover completed successfully");  
        } catch (Exception e) {  
            log.error("Failover failed", e);  
            alertManager.sendCriticalAlert("Failover failed: " + e.getMessage());  
        }  
    }  
}

// 性能測(cè)試數(shù)據(jù)生成器  
public class TransactionDataGenerator extends RichSourceFunction<Transaction> {  
    private volatile boolean isRunning = true;  
    private final int transactionsPerSecond;  
    private final Random random = new Random();  


    public TransactionDataGenerator(int transactionsPerSecond) {  
        this.transactionsPerSecond = transactionsPerSecond;  
    }  


    @Override  
    public void run(SourceContext<Transaction> ctx) throws Exception {  
        long intervalMs = 1000 / transactionsPerSecond;  


        while (isRunning) {  
            Transaction transaction = generateRandomTransaction();  
            ctx.collect(transaction);  
            Thread.sleep(intervalMs);  
        }  
    }  


    private Transaction generateRandomTransaction() {  
        return Transaction.builder()  
            .transactionId(System.currentTimeMillis() + random.nextInt(1000))  
            .userId(random.nextLong() % 1000000) // 100萬(wàn)用戶  
            .amount(BigDecimal.valueOf(random.nextDouble() * 10000)) // 0-10000金額  
            .merchantId(random.nextLong() % 10000) // 1萬(wàn)商戶  
            .transactionType(getRandomTransactionType())  
            .transactionTime(Timestamp.valueOf(LocalDateTime.now()))  
            .ipAddress(generateRandomIp())  
            .deviceId(generateRandomDeviceId())  
            .location(getRandomLocation())  
            .build();  
    }  
}

-- 性能基準(zhǔn)測(cè)試結(jié)果表  
CREATE TABLE performance_benchmark (  
    test_id STRING,  
    test_timestamp TIMESTAMP(3),  
    transactions_per_second BIGINT,  
    avg_latency_ms BIGINT,  
    p95_latency_ms BIGINT,  
    p99_latency_ms BIGINT,  
    cpu_utilization DOUBLE,  
    memory_utilization DOUBLE,  
    throughput_mbps DOUBLE,  
    error_rate DOUBLE,  
    test_duration_minutes INT,  
    cluster_size INT,  
    parallelism INT  
) WITH (  
    'connector' = 'jdbc',  
    'url' = 'jdbc:mysql://localhost:3306/performance',  
    'table-name' = 'benchmark_results'  
);