AI預(yù)測(cè)：基于流計(jì)算Oceanus (Flink) 實(shí)現(xiàn)病癥的實(shí)時(shí)預(yù)測(cè)

# 下載 ClickHouse-Client 命令wget https://repo.yandex.ru/clickhouse/rpm/stable/x86_64/clickhouse-client-20.7.2.30-2.noarch.rpmwget https://repo.yandex.ru/clickhouse/rpm/stable/x86_64/clickhouse-common-static-20.7.2.30-2.x86_64.rpm
# 安裝客戶端rpm -ivh *.rpm
# 使用 tcp 端口登陸 ClickHouse 集群，IP 地址可通過(guò)控制臺(tái)查看clickhouse-client -hxxx.xxx.xxx.xxx --port 9000

-- 創(chuàng)建數(shù)據(jù)庫(kù)CREATE DATABASE IF NOT EXISTS testdb ON CLUSTER default_cluster;
-- 創(chuàng)建表CREATE TABLE testdb.model_predict_result_1 on cluster default_cluster (res String,Sign Int8) ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/testdb/model_predict_result_1', '{replica}',Sign) ORDER BY res;

-- random source 用于模擬患者病歷實(shí)時(shí)特征數(shù)據(jù)
CREATE TABLE random_source (     ClumpThickness             INT,    UniformityOfCellSize       INT,    UniformityOfCellShape      INT,    MarginalAdhsion            INT,    SingleEpithelialCellSize   INT,    BareNuclei                 INT,    BlandChromation            INT,    NormalNucleoli             INT,    Mitoses                    INT  ) WITH (  'connector' = 'datagen',   'rows-per-second'='1',                           -- 每秒產(chǎn)生的數(shù)據(jù)條數(shù)  'fields.ClumpThickness.kind'='random',           -- 無(wú)界的隨機(jī)數(shù)  'fields.ClumpThickness.min'='0',                 -- 隨機(jī)數(shù)的最小值  'fields.ClumpThickness.max'='10',                -- 隨機(jī)數(shù)的最大值  'fields.UniformityOfCellSize.kind'='random',     -- 無(wú)界的隨機(jī)數(shù)  'fields.UniformityOfCellSize.min'='0',           -- 隨機(jī)數(shù)的最小值  'fields.UniformityOfCellSize.max'='10',          -- 隨機(jī)數(shù)的最大值  'fields.UniformityOfCellShape.kind'='random',    -- 無(wú)界的隨機(jī)數(shù)  'fields.UniformityOfCellShape.min'='0',          -- 隨機(jī)數(shù)的最小值  'fields.UniformityOfCellShape.max'='10',         -- 隨機(jī)數(shù)的最大值  'fields.MarginalAdhsion.kind'='random',          -- 無(wú)界的隨機(jī)數(shù)  'fields.MarginalAdhsion.min'='0',                -- 隨機(jī)數(shù)的最小值  'fields.MarginalAdhsion.max'='10',               -- 隨機(jī)數(shù)的最大值  'fields.SingleEpithelialCellSize.kind'='random', -- 無(wú)界的隨機(jī)數(shù)  'fields.SingleEpithelialCellSize.min'='0',       -- 隨機(jī)數(shù)的最小值  'fields.SingleEpithelialCellSize.max'='10',      -- 隨機(jī)數(shù)的最大值  'fields.BareNuclei.kind'='random',               -- 無(wú)界的隨機(jī)數(shù)  'fields.BareNuclei.min'='0',                     -- 隨機(jī)數(shù)的最小值  'fields.BareNuclei.max'='10',                    -- 隨機(jī)數(shù)的最大值  'fields.BlandChromation.kind'='random',          -- 無(wú)界的隨機(jī)數(shù)  'fields.BlandChromation.min'='0',                -- 隨機(jī)數(shù)的最小值  'fields.BlandChromation.max'='10',               -- 隨機(jī)數(shù)的最大值  'fields.NormalNucleoli.kind'='random',           -- 無(wú)界的隨機(jī)數(shù)  'fields.NormalNucleoli.min'='0',                 -- 隨機(jī)數(shù)的最小值  'fields.NormalNucleoli.max'='10',                -- 隨機(jī)數(shù)的最大值  'fields.Mitoses.kind'='random',                  -- 無(wú)界的隨機(jī)數(shù)  'fields.Mitoses.min'='0',                        -- 隨機(jī)數(shù)的最小值  'fields.Mitoses.max'='10'                        -- 隨機(jī)數(shù)的最大值);

 CREATE TABLE `KafkaSink` (    ClumpThickness             INT,    UniformityOfCellSize       INT,    UniformityOfCellShape      INT,    MarginalAdhsion            INT,    SingleEpithelialCellSize   INT,    BareNuclei                 INT,    BlandChromation            INT,    NormalNucleoli             INT,    Mitoses                    INT ) WITH (     'connector' = 'kafka',                                  -- 可選 'kafka','kafka-0.11'. 注意選擇對(duì)應(yīng)的內(nèi)置  Connector     'topic' = 'topic-decision-tree-predict-1',              -- 替換為您要消費(fèi)的 Topic     'properties.bootstrap.servers' = '172.28.28.211:9092',  -- 替換為您的 Kafka 連接地址     'properties.group.id' = 'RealTimeFeatures',             -- 必選參數(shù), 一定要指定 Group ID     'format' = 'csv' );

 INSERT INTO `KafkaSink` SELECT * FROM `random_source`

# 請(qǐng)將 <訪問(wèn)地址>/<密鑰> 替換為實(shí)際的 IP 地址/密鑰curl -H "Content-Type: application/json" \-H "x-Auth-Token: <密鑰>" \-X POST <訪問(wèn)地址>/v1/models/m:predict -d @data.json

{"instances" : [{"_c0": 3, "_c1": 2, "_c2": 3, "_c3": 0, "_c4": 0, "_c5": 2, "_c6": 1, "_c7": 0, "_c8": 1}]}

{"predictions": [{"pmml(prediction)":"1","probability(0)":"0.47058823529411764","probability(1)":"0.5294117647058824","prediction":"1.0","label":"1"}]}

import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.restartstrategy.RestartStrategies;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.api.java.utils.ParameterTool;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.CheckpointConfig;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import org.apache.flink.table.api.Table;import org.apache.flink.util.Collector;import org.apache.http.HttpEntity;import org.apache.http.HttpResponse;import org.apache.http.client.HttpClient;import org.apache.http.client.methods.HttpPost;import org.apache.http.entity.StringEntity;import org.apache.http.impl.client.HttpClientBuilder;import org.apache.http.util.EntityUtils;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.json.JSONObject;import org.slf4j.LoggerFactory;import org.slf4j.Logger;import java.util.ArrayList;import java.util.Properties;import org.apache.flink.table.api.EnvironmentSettings;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class OnlinePredict {
    public static final Logger logger = LoggerFactory.getLogger(OnlinePredict.class);
    public static void main(String[] args) throws Exception {        // kafka配置參數(shù)解析        final ParameterTool parameterTool = ParameterTool                .fromPropertiesFile(OnlinePredict.class.getResourceAsStream("/KafkaSource.properties"));        // 實(shí)例化運(yùn)行環(huán)境        EnvironmentSettings settings = EnvironmentSettings                .newInstance()                .useBlinkPlanner()                .inStreamingMode()                .build();        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(streamEnv, settings);
        // checkpoint配置        streamEnv.enableCheckpointing(parameterTool.getLong("flink.stream.checkpoint.interval", 30_000));        streamEnv.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);        // 重啟策略        streamEnv.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10_000));        // source、transfer、sink        DataStream stringResult = streamEnv.addSource(buildKafkaSource(parameterTool))                .flatMap(new FlatMapFunction() {                    @Override                    public void flatMap(String value, Collector out) throws Exception {                        String paramInput = inputDataTransfer(value);                        String outputData = sendHttpData(paramInput);                        out.collect(outputData);                    }                });
        Table tableResult = tableEnv.fromDataStream(stringResult);        tableEnv.createTemporaryView("resultSink",tableResult);
        tableEnv.executeSql("CREATE TABLE `CKSink` (\n" +                "  res STRING,\n" +                "  PRIMARY KEY (`res`) NOT ENFORCED\n" +                ") WITH (\n" +                "  'connector' = 'clickhouse',\n" +                "  'url' = 'clickhouse://172.28.1.138:8123',\n" +                "  'database-name' = 'testdb',\n" +                "  'table-name' = 'model_predict_result_1',\n" +                "  'table.collapsing.field' = 'Sign'\n" +                ")");
        tableEnv.executeSql("insert into CKSink select * from resultSink");
    }
    // kafka source    public static SourceFunction buildKafkaSource(ParameterTool parameterTool) throws Exception {        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, parameterTool.get("kafka.source.bootstrap.servers"));        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, parameterTool.get("kafka.source.auto.offset.reset", "latest"));        properties.put(ConsumerConfig.GROUP_ID_CONFIG, parameterTool.get("kafka.source.group.id"));
        FlinkKafkaConsumer consumer = new FlinkKafkaConsumer(                parameterTool.get("kafka.source.topic"),                new SimpleStringSchema(),                properties);        consumer.setStartFromGroupOffsets();
        return consumer;    }
    // kafka 數(shù)據(jù)格式轉(zhuǎn)換    // 返回?cái)?shù)據(jù)格式：{"instances" : [{"_c0": 3, "_c1": 2, "_c2": 3, "_c3": 0, "_c4": 0, "_c5": 2, "_c6": 1, "_c7": 0, "_c8": 1}]}    public static String inputDataTransfer(String value) {        String[] input = value.split(",");        ArrayList dataListMap = new ArrayList();        JSONObject jsondata = new JSONObject();        for (int i = 0; i < input.length; i++) {            jsondata.put("_c" + i, Double.parseDouble(input[i]));        }        dataListMap.add(jsondata);        String param = "{\"instances\":" + dataListMap.toString() + "}";        return param;    }
    // TI-EMS 模型在線推理服務(wù)調(diào)用    // 返回?cái)?shù)據(jù)格式如下：{"predictions": [{"pmml(prediction)":"1","probability(0)":"0.47058823529411764","probability(1)":"0.5294117647058824","prediction":"1.0","label":"1"}]}    public static String sendHttpData(String paramJson) throws Exception {        String data = null;        try {            // 請(qǐng)將 xx.xx.xx.xx:xxxx 替換為實(shí)際的 IP 地址，參考 3.2.2 圖中所示 創(chuàng)建 VPC 調(diào)用            String url = "http://xx.xx.xx.xx:xxxx/v1/models/m:predict";             HttpClient client = HttpClientBuilder.create().build();            HttpPost post = new HttpPost(url);
            post.addHeader("Content-type", "application/json");            post.addHeader("Accept", "application/json");            // 請(qǐng)將 xxxxxxxxxx 替換為實(shí)際密鑰，參考 3.2.2 圖中所示 創(chuàng)建 VPC 調(diào)用            post.addHeader("X-AUTH-TOKEN", "xxxxxxxxxx");   
            StringEntity entity = new StringEntity(paramJson, java.nio.charset.Charset.forName("UTF-8"));            post.setEntity(entity);            HttpResponse response = client.execute(post);
            // 判斷是否正常返回            if (response.getStatusLine().getStatusCode() == 200) {                // 解析數(shù)據(jù)                HttpEntity resEntity = response.getEntity();                data = EntityUtils.toString(resEntity);            } else {                data = "error input";            }            System.out.print(data);            System.out.println(data);        } catch (Throwable e) {            logger.error("", e);        }        return data;    }
}

# source // 請(qǐng)?zhí)鎿Q為實(shí)際的參數(shù)kafka.source.bootstrap.servers=172.28.28.211:9092kafka.source.topic=topic-decision-tree-predict-1kafka.source.group.id=RealTimePredict1kafka.source.auto.offset.reset=latest

<properties>    <flink.version>1.11.0flink.version>properties>
<dependencies>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-streaming-java_2.11artifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-connector-kafka_2.11artifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-connector-clickhouseartifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-table-commonartifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-table-api-java-bridge_2.11artifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
        <dependency>            <groupId>org.apache.flinkgroupId>            <artifactId>flink-table-api-javaartifactId>            <version>${flink.version}version>            <scope>providedscope>        dependency>
                <dependency>            <groupId>org.apache.httpcomponentsgroupId>            <artifactId>httpclientartifactId>            <version>4.5.3version>            <scope>compilescope>        dependency>
        <dependency>            <groupId>org.jsongroupId>            <artifactId>jsonartifactId>            <version>20201115version>            <scope>compilescope>        dependency>
dependencies>