拥抱AI-实时特征工程 + 轻量模型 —— 实时管道的智能化

Kafka → Flink ETL → 特征计算 → 轻量模型推理 → 实时结果写入                              ↑                        模型从云存储加载

Flink TaskManager JVM├── Task线程│   ├── 数据反序列化│   ├── 特征提取│   └── ONNX Runtime推理 ← 在这里│       ├── 加载模型文件（.onnx）│       └── 执行推理（CPU）└── 共享资源    └── ORT会话池（每个subtask一个）

生成Isolation Forest模型的训练代码（Python）：数据：transactions.parquet（1个月历史交易）特征列：- amount_zscore: double（金额偏离度）- tx_count_5min: int（5分钟交易次数）- is_remote_location: int（0/1）- device_risk_score: double（设备历史风险分）标签：is_fraud（0/1，用于验证）要求：1. 使用sklearn.ensemble.IsolationForest2. 设置contamination=0.01（1%异常）3. 训练后保存模型为ONNX格式4. 输出：模型文件、特征重要性、阈值生成完整Python脚本。

# train_isolation_forest.pyimport pandas as pdimport numpy as npfrom sklearn.ensemble import IsolationForestfrom sklearn.metrics import classification_reportimport onnximport skl2onnxfrom skl2onnx import convert_sklearnfrom skl2onnx.common.data_types import FloatTensorType# 1. 加载数据df = pd.read_parquet('transactions.parquet')features = ['amount_zscore', 'tx_count_5min', 'is_remote_location', 'device_risk_score']X = df[features].valuesy = df['is_fraud'].values# 2. 训练模型model = IsolationForest(    n_estimators=100,    max_samples='auto',    contamination=0.01,    random_state=42,    n_jobs=-1)model.fit(X)# 3. 预测（-1是异常，1是正常）y_pred = model.predict(X)y_pred_binary = np.where(y_pred == -1, 1, 0)# 4. 评估print(classification_report(y, y_pred_binary))# 5. 确定阈值（用原始异常分数）scores = model.score_samples(X)threshold = np.percentile(scores, 1)  # 1%分位数作为阈值print(f"建议阈值: {threshold}")# 6. 转换为ONNXinitial_type = [('float_input', FloatTensorType([None, len(features)]))]onnx_model = convert_sklearn(model, initial_types=initial_type)# 7. 保存with open("isolation_forest.onnx", "wb") as f:    f.write(onnx_model.SerializeToString())print("模型已保存为 isolation_forest.onnx")

在Flink DataStream API中部署ONNX模型进行实时推理：模型：isolation_forest.onnx（输入4个特征，输出异常分数）输入流：TransactionEvent- userId: String- amount: Double- merchant: String- location: String- deviceId: String- timestamp: Long处理流程：1. 实时特征计算（基于Flink状态）：   - 用户历史平均金额（ValueState存储金额序列，5分钟滑动窗口）   - amount_zscore = (当前金额 - 平均金额) / 标准差   - 过去5分钟交易次数（滑动窗口计数）   - 是否异地（当前location != 历史常驻地）   - 设备风险分（从Redis维表查询）2. 特征组装成float数组3. ONNX Runtime推理，得到异常分数4. 如果分数超过阈值，输出告警到侧输出流要求：- 使用ONNX Runtime Java API（onnxruntime 1.15+）- 每个subtask共享一个ORT会话（避免重复加载）- 特征计算使用Flink状态（ValueState + MapState）- 异常阈值可配置- 必须设置状态TTL生成完整的KeyedProcessFunction代码。

package com.company.realtime.anomaly;import ai.onnxruntime.*;import org.apache.flink.api.common.state.*;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.KeyedProcessFunction;import org.apache.flink.util.Collector;import java.io.File;import java.util.*;/** * 实时异常检测ProcessFunction */public class AnomalyDetectionFunction     extends KeyedProcessFunction<String, TransactionEvent, TransactionEvent> {    // ONNX Runtime相关    private transient OrtEnvironment env;    private transient OrtSession session;    // 状态：用户历史金额列表（保留最近5分钟）    private transient MapState<Long, Double> amountHistoryState;    // 状态：用户常驻地    private transient ValueState<String> usualLocationState;    // 配置    private final String modelPath;    private final double anomalyThreshold;    private final int windowSeconds = 300; // 5分钟    public AnomalyDetectionFunction(String modelPath, double anomalyThreshold) {        this.modelPath = modelPath;        this.anomalyThreshold = anomalyThreshold;    }    @Override    public void open(Configuration parameters) throws Exception {        super.open(parameters);        // 1. 初始化ONNX Runtime        env = OrtEnvironment.getEnvironment();        OrtSession.SessionOptions sessionOptions = new OrtSession.SessionOptions();        sessionOptions.setOptimizationLevel(OrtSession.SessionOptions.OptLevel.ALL_OPT);        session = env.createSession(modelPath, sessionOptions);        // 2. 初始化状态        MapStateDescriptor<Long, Double> amountDescriptor =             new MapStateDescriptor<>("amount-history", Long.class, Double.class);        StateTtlConfig ttlConfig = StateTtlConfig            .newBuilder(org.apache.flink.api.common.time.Time.seconds(windowSeconds + 60))            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)            .build();        amountDescriptor.enableTimeToLive(ttlConfig);        amountHistoryState = getRuntimeContext().getMapState(amountDescriptor);        ValueStateDescriptor<String> locationDescriptor =             new ValueStateDescriptor<>("usual-location", String.class);        usualLocationState = getRuntimeContext().getState(locationDescriptor);    }    @Override    public void processElement(            TransactionEvent event,            Context ctx,            Collector<TransactionEvent> out) throws Exception {        long currentTime = event.getTimestamp();        long windowStart = currentTime - windowSeconds * 1000;        // 1. 清理过期金额数据        cleanupExpiredAmounts(windowStart);        // 2. 计算特征        float[] features = extractFeatures(event, windowStart);        // 3. ONNX推理        double anomalyScore = predict(features);        // 4. 判断异常        if (anomalyScore < anomalyThreshold) {            // 异常，输出到侧输出流            ctx.output(AnomalySideOutputs.ALERT_TAG, event);        } else {            // 正常            out.collect(event);        }        // 5. 更新状态        amountHistoryState.put(currentTime, event.getAmount());        updateUsualLocation(event.getLocation());    }    /**     * 提取特征     */    private float[] extractFeatures(TransactionEvent event, long windowStart) throws Exception {        // 特征1: amount_zscore        double amountZscore = calculateZscore(event.getAmount(), windowStart);        // 特征2: 过去5分钟交易次数        int txCount = getTxCountInWindow(windowStart);        // 特征3: 是否异地        int isRemote = isRemoteLocation(event.getLocation()) ? 1 : 0;        // 特征4: 设备风险分（从Redis或其他来源）        double deviceRiskScore = getDeviceRiskScore(event.getDeviceId());        return new float[]{            (float) amountZscore,            (float) txCount,            (float) isRemote,            (float) deviceRiskScore        };    }    /**     * 计算金额z-score     */    private double calculateZscore(double currentAmount, long windowStart) throws Exception {        List<Double> amounts = new ArrayList<>();        for (Map.Entry<Long, Double> entry : amountHistoryState.entries()) {            if (entry.getKey() >= windowStart) {                amounts.add(entry.getValue());            }        }        if (amounts.isEmpty()) {            return 0.0; // 无历史数据        }        double mean = amounts.stream().mapToDouble(Double::doubleValue).average().orElse(0);        double std = Math.sqrt(amounts.stream()            .mapToDouble(a -> Math.pow(a - mean, 2))            .average().orElse(0));        if (std < 0.01) {            return 0.0; // 标准差太小，认为正常        }        return (currentAmount - mean) / std;    }    /**     * ONNX推理     */    private double predict(float[] features) throws OrtException {        // 创建输入张量        OnnxTensor input = OnnxTensor.createTensor(env, new float[][]{features});        // 运行推理        OrtSession.Result result = session.run(Collections.singletonMap("float_input", input));        // 获取输出（异常分数）        OnnxTensor output = (OnnxTensor) result.get(0);        float[][] scores = (float[][]) output.getValue();        result.close();        input.close();        return scores[0][0]; // Isolation Forest的分数    }    private void cleanupExpiredAmounts(long windowStart) throws Exception {        Iterator<Map.Entry<Long, Double>> it = amountHistoryState.iterator();        while (it.hasNext()) {            Map.Entry<Long, Double> entry = it.next();            if (entry.getKey() < windowStart) {                it.remove();            }        }    }    private int getTxCountInWindow(long windowStart) throws Exception {        int count = 0;        for (Map.Entry<Long, Double> entry : amountHistoryState.entries()) {            if (entry.getKey() >= windowStart) {                count++;            }        }        return count;    }    private boolean isRemoteLocation(String currentLocation) throws Exception {        String usual = usualLocationState.value();        if (usual == null) {            return false; // 首次交易，不确定        }        return !usual.equals(currentLocation);    }    private void updateUsualLocation(String location) throws Exception {        String currentUsual = usualLocationState.value();        if (currentUsual == null) {            usualLocationState.update(location);        }        // 这里可以加更复杂的逻辑：比如多数投票    }    private double getDeviceRiskScore(String deviceId) {        // 从Redis或其他维表查询        // 简化示例，实际应该异步I/O        return 0.5;    }    @Override    public void close() throws Exception {        if (session != null) {            session.close();        }        if (env != null) {            env.close();        }        super.close();    }}// 侧输出Tag定义class AnomalySideOutputs {    static final org.apache.flink.util.OutputTag<TransactionEvent> ALERT_TAG =        new org.apache.flink.util.OutputTag<TransactionEvent>("anomaly-alert") {};}

生成将DistilBERT转换为ONNX的脚本（Python）：要求：1. 使用transformers库加载distilbert-base-uncased-finetuned-sst-2-english2. 转换为ONNX格式，支持动态序列长度3. 使用量化（INT8）减小模型大小4. 输出：model.onnx, tokenizer_config.json, vocab.txt生成完整Python脚本。

# convert_bert_to_onnx.pyfrom transformers import DistilBertTokenizer, DistilBertForSequenceClassificationimport torchimport onnxfrom onnxruntime.quantization import quantize_dynamic, QuantType# 1. 加载模型model_name = "distilbert-base-uncased-finetuned-sst-2-english"tokenizer = DistilBertTokenizer.from_pretrained(model_name)model = DistilBertForSequenceClassification.from_pretrained(model_name)# 2. 设置模型为eval模式model.eval()# 3. 创建虚拟输入（动态序列长度）dummy_input = tokenizer(    "This is a test sentence",    return_tensors="pt",    padding="max_length",    max_length=128,    truncation=True)# 4. 导出ONNXtorch.onnx.export(    model,    (dummy_input["input_ids"], dummy_input["attention_mask"]),    "bert_sentiment.onnx",    input_names=["input_ids", "attention_mask"],    output_names=["logits"],    dynamic_axes={        "input_ids": {0: "batch_size", 1: "sequence_length"},        "attention_mask": {0: "batch_size", 1: "sequence_length"},        "logits": {0: "batch_size"}    },    opset_version=14)# 5. 量化模型（减小大小）quantize_dynamic(    "bert_sentiment.onnx",    "bert_sentiment_int8.onnx",    weight_type=QuantType.QInt8)# 6. 保存tokenizertokenizer.save_pretrained("./bert_tokenizer")print("模型转换完成！")print("原始大小: ~250MB")print("量化后大小: ~50MB")

在Flink中部署BERT文本分类：挑战：- 模型大小50MB，每个subtask加载一次 → 总内存 16×50MB=800MB（可接受）- 文本需要tokenization，JVM中实现较复杂解决方案：1. 在Flink中预加载tokenizer（Python tokenizer不能直接用）2. 使用Java版tokenizer（如HuggingFace Tokenizer的Java绑定）3. 或者：把tokenizer也转成ONNX的一部分（不推荐）替代方案：预处理tokenization在Flink中完成，然后传给ONNX模型。请生成：1. Java端tokenizer的实现（基于org.apache.lucene.analysis）2. 推理的KeyedProcessFunction3. 批量推理优化（攒batch）注意：如果模型太大，考虑用外部推理服务。

优化推理性能，实现批量推理：在KeyedProcessFunction中增加buffer，攒够batchSize=32或超过timeout=100ms触发推理。要求：- 使用ListState缓存输入- 定时器触发flush- 推理时一次性输入batchSize条数据- 输出时按顺序返回结果生成优化后的代码。

public class BatchAnomalyDetectionFunction     extends KeyedProcessFunction<String, TransactionEvent, TransactionEvent> {    private transient ListState<TransactionEvent> bufferState;    private transient ValueState<Long> timerState;    private final int batchSize = 32;    private final long batchTimeoutMs = 100;    @Override    public void processElement(TransactionEvent event, Context ctx, Collector<TransactionEvent> out)             throws Exception {        // 添加到buffer        bufferState.add(event);        // 注册或更新定时器        Long currentTimer = timerState.value();        if (currentTimer == null) {            long triggerTime = ctx.timerService().currentProcessingTime() + batchTimeoutMs;            ctx.timerService().registerProcessingTimeTimer(triggerTime);            timerState.update(triggerTime);        }        // 如果达到batch大小，立即触发        Iterable<TransactionEvent> buffer = bufferState.get();        int size = 0;        for (TransactionEvent e : buffer) size++;        if (size >= batchSize) {            // 取消定时器，立即处理            if (currentTimer != null) {                ctx.timerService().deleteProcessingTimeTimer(currentTimer);            }            processBatch(ctx, out);        }    }    @Override    public void onTimer(long timestamp, OnTimerContext ctx, Collector<TransactionEvent> out)             throws Exception {        processBatch(ctx, out);    }    private void processBatch(OnTimerContext ctx, Collector<TransactionEvent> out) throws Exception {        List<TransactionEvent> batch = new ArrayList<>();        bufferState.get().forEach(batch::add);        if (batch.isEmpty()) return;        // 批量推理        float[][] features = batch.stream()            .map(this::extractFeatures)            .toArray(float[][]::new);        double[] scores = batchPredict(features);        // 输出结果        for (int i = 0; i < batch.size(); i++) {            if (scores[i] < anomalyThreshold) {                ctx.output(AnomalySideOutputs.ALERT_TAG, batch.get(i));            } else {                out.collect(batch.get(i));            }        }        // 清理状态        bufferState.clear();        timerState.clear();    }    private double[] batchPredict(float[][] features) throws OrtException {        OnnxTensor input = OnnxTensor.createTensor(env, features);        OrtSession.Result result = session.run(Collections.singletonMap("float_input", input));        OnnxTensor output = (OnnxTensor) result.get(0);        float[][] scores = (float[][]) output.getValue();        result.close();        input.close();        return Arrays.stream(scores).mapToDouble(s -> s[0]).toArray();    }}

@Overridepublic void open(Configuration parameters) throws Exception {    // ... 初始化    // 预热：执行一次dummy推理    float[] dummyFeatures = new float[]{0, 0, 0, 0};    predict(dummyFeatures);}

public class SharedModelLoader {    private static OrtSession sharedSession;    private static final Object lock = new Object();    public static OrtSession getSession(String modelPath) throws OrtException {        if (sharedSession == null) {            synchronized (lock) {                if (sharedSession == null) {                    OrtEnvironment env = OrtEnvironment.getEnvironment();                    sharedSession = env.createSession(modelPath);                }            }        }        return sharedSession;    }}

实现模型热加载：定期从HDFS检查模型文件是否更新，如果更新则重新加载。要求：- 每5分钟检查一次模型文件的MD5- 如果变化，重新加载ONNX Session- 旧Session优雅关闭（等待当前推理完成）- 使用原子引用保证线程安全生成代码。

public class HotLoadableModel {    private AtomicReference<OrtSession> currentSession = new AtomicReference<>();    private String currentModelMd5;    private final String modelPath;    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();    public HotLoadableModel(String modelPath) {        this.modelPath = modelPath;        loadModel();        startPeriodicCheck();    }    private void loadModel() {        try {            String newMd5 = calculateMd5(modelPath);            if (newMd5.equals(currentModelMd5)) {                return; // 无变化            }            OrtEnvironment env = OrtEnvironment.getEnvironment();            OrtSession newSession = env.createSession(modelPath);            // 原子替换            OrtSession oldSession = currentSession.getAndSet(newSession);            currentModelMd5 = newMd5;            // 延迟关闭旧session（等待使用中的推理完成）            if (oldSession != null) {                scheduler.schedule(oldSession::close, 10, TimeUnit.SECONDS);            }            System.out.println("Model reloaded: " + modelPath);        } catch (Exception e) {            System.err.println("Failed to reload model: " + e.getMessage());        }    }    private void startPeriodicCheck() {        scheduler.scheduleAtFixedRate(this::loadModel, 5, 5, TimeUnit.MINUTES);    }    public OrtSession getSession() {        return currentSession.get();    }    private String calculateMd5(String path) throws IOException {        try (InputStream is = new FileInputStream(path)) {            return org.apache.commons.codec.digest.DigestUtils.md5Hex(is);        }    }}

实现模型A/B测试：同时加载模型A和模型B，10%流量走模型B。要求：- 根据user_id hash决定走哪个模型- 分别记录两个模型的推理耗时和准确率- 输出到不同的metrics生成代码。

Kafka(交易事件)     ↓Flink Job: RiskScoringJob    ├── 特征计算（状态）    │   ├── 用户画像（金额、频次、设备）    │   ├── 地理特征（IP、定位）    │   └── 时间特征（时段、节假日）    ├── 模型1: Isolation Forest（无监督）    ├── 模型2: XGBoost（有监督）    └── 模型融合（加权平均）    ↓输出：  ├── 高风险 → Kafka告警Topic  ├── 中风险 → MySQL（待人工审核）  └── 低风险 → Redis（实时通过）

生成实时风控评分系统的完整代码：【数据源】- Kafka: transactions（交易事件）- Redis: user_risk_history（用户历史风险分）- MySQL: blacklist_device（黑名单设备）【特征】（18个）- 金额特征：amount, amount_zscore, amount_std_5min- 频次特征：tx_count_5min, tx_count_1hour, tx_count_1day- 设备特征：is_emulator, is_rooted, device_risk_score- 地理特征：is_remote_location, distance_from_usual- 时间特征：is_night, is_weekend, days_since_first_tx- 行为特征：is_fast_tx（<10秒内多次交易）【模型】- Model A: Isolation Forest（异常检测）- Model B: XGBoost（二分类，评分0-100）- 融合：risk_score = 0.3 * anomaly_score + 0.7 * xgboost_score【输出】- risk_score < 30: LOW → 直接通过- 30 ≤ risk_score < 70: MEDIUM → 存入MySQL待审核- risk_score ≥ 70: HIGH → 发送Kafka告警要求：1. 使用Flink DataStream API2. 所有特征用状态计算3. 模型用ONNX Runtime加载4. 输出到多个Sink5. 必须有完整的单元测试生成所有Java类和配置文件。