Java 类名：com.alibaba.alink.operator.stream.clustering.GmmPredictStreamOp
Python 类名：GmmPredictStreamOp

功能介绍

基于GaussianMixture模型进行聚类预测。

参数说明

| 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| predictionCol | 预测结果列名 | 预测结果列名 | String | ✓ | | |

| vectorCol | 向量列名 | 向量列对应的列名 | String | ✓ | 所选列类型为 [DENSE_VECTOR, SPARSE_VECTOR, STRING, VECTOR] | |

| modelFilePath | 模型的文件路径 | 模型的文件路径 | String | | | null |

| predictionDetailCol | 预测详细信息列名 | 预测详细信息列名 | String | | | |

| reservedCols | 算法保留列名 | 算法保留列 | String[] | | | null |

| numThreads | 组件多线程线程个数 | 组件多线程线程个数 | Integer | | | 1 |

| modelStreamFilePath | 模型流的文件路径 | 模型流的文件路径 | String | | | null |

| modelStreamScanInterval | 扫描模型路径的时间间隔 | 描模型路径的时间间隔，单位秒 | Integer | | | 10 |

| modelStreamStartTime | 模型流的起始时间 | 模型流的起始时间。默认从当前时刻开始读。使用yyyy-mm-dd hh:mm:ss.fffffffff格式，详见Timestamp.valueOf(String s) | String | | | null |

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. df_data = pd.DataFrame([
  8.     ["-0.6264538 0.1836433"],
  9.     ["-0.8356286 1.5952808"],
  10.     ["0.3295078 -0.8204684"],
  11.     ["0.4874291 0.7383247"],
  12.     ["0.5757814 -0.3053884"],
  13.     ["1.5117812 0.3898432"],
  14.     ["-0.6212406 -2.2146999"],
  15.     ["11.1249309 9.9550664"],
  16.     ["9.9838097 10.9438362"],
  17.     ["10.8212212 10.5939013"],
  18.     ["10.9189774 10.7821363"],
  19.     ["10.0745650 8.0106483"],
  20.     ["10.6198257 9.9438713"],
  21.     ["9.8442045 8.5292476"],
  22.     ["9.5218499 10.4179416"],
  23. ])
  25. data = BatchOperator.fromDataframe(df_data, schemaStr='features string')
  26. dataStream = StreamOperator.fromDataframe(df_data, schemaStr='features string')
  28. gmm = GmmTrainBatchOp() \
  29.     .setVectorCol("features") \
  30.     .setEpsilon(0.)
  32. model = gmm.linkFrom(data)
  34. predictor = GmmPredictBatchOp() \
  35.     .setPredictionCol("cluster_id") \
  36.     .setVectorCol("features") \
  37.     .setPredictionDetailCol("cluster_detail")
  39. predictor.linkFrom(model, data).print()
  41. predictorStream = GmmPredictStreamOp(model) \
  42.     .setPredictionCol("cluster_id") \
  43.     .setVectorCol("features") \
  44.     .setPredictionDetailCol("cluster_detail")
  46. predictorStream.linkFrom(dataStream).print()
  48. StreamOperator.execute()

Java 代码

  1. import org.apache.flink.types.Row;
  3. import com.alibaba.alink.operator.batch.BatchOperator;
  4. import com.alibaba.alink.operator.batch.clustering.GmmPredictBatchOp;
  5. import com.alibaba.alink.operator.batch.clustering.GmmTrainBatchOp;
  6. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  7. import com.alibaba.alink.operator.stream.StreamOperator;
  8. import com.alibaba.alink.operator.stream.clustering.GmmPredictStreamOp;
  9. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  10. import org.junit.Test;
  12. import java.util.Arrays;
  13. import java.util.List;
  15. public class GmmPredictStreamOpTest {
  16.     @Test
  17.     public void testGmmPredictStreamOp() throws Exception {
  18.         List <Row> df_data = Arrays.asList(
  19.             Row.of("-0.6264538 0.1836433"),
  20.             Row.of("-0.8356286 1.5952808"),
  21.             Row.of("0.3295078 -0.8204684"),
  22.             Row.of("0.4874291 0.7383247"),
  23.             Row.of("0.5757814 -0.3053884"),
  24.             Row.of("1.5117812 0.3898432"),
  25.             Row.of("-0.6212406 -2.2146999"),
  26.             Row.of("11.1249309 9.9550664"),
  27.             Row.of("9.9838097 10.9438362"),
  28.             Row.of("10.8212212 10.5939013"),
  29.             Row.of("10.9189774 10.7821363"),
  30.             Row.of("10.0745650 8.0106483"),
  31.             Row.of("10.6198257 9.9438713"),
  32.             Row.of("9.8442045 8.5292476"),
  33.             Row.of("9.5218499 10.4179416")
  34.         );
  35.         BatchOperator <?> data = new MemSourceBatchOp(df_data, "features string");
  36.         StreamOperator <?> dataStream = new MemSourceStreamOp(df_data, "features string");
  37.         BatchOperator <?> gmm = new GmmTrainBatchOp()
  38.             .setVectorCol("features")
  39.             .setEpsilon(0.);
  40.         BatchOperator <?> model = gmm.linkFrom(data);
  41.         BatchOperator <?> predictor = new GmmPredictBatchOp()
  42.             .setPredictionCol("cluster_id")
  43.             .setVectorCol("features")
  44.             .setPredictionDetailCol("cluster_detail");
  45.         predictor.linkFrom(model, data).print();
  46.         StreamOperator <?> predictorStream = new GmmPredictStreamOp(model)
  47.             .setPredictionCol("cluster_id")
  48.             .setVectorCol("features")
  49.             .setPredictionDetailCol("cluster_detail");
  50.         predictorStream.linkFrom(dataStream).print();
  51.         StreamOperator.execute();
  52.     }
  53. }

运行结果

| features | cluster_id | cluster_detail | | —- | —- | —- |

| -0.6264538 0.1836433 | 1 | 4.2752739140319505E-92 1.0 |

| -0.8356286 1.5952808 | 1 | 1.0260377730418951E-92 1.0 |

| 0.3295078 -0.8204684 | 1 | 1.097017336765683E-80 1.0 |

| 0.4874291 0.7383247 | 1 | 3.3021731323490106E-75 1.0 |

| 0.5757814 -0.3053884 | 1 | 3.163811360548103E-76 1.0 |

| 1.5117812 0.3898432 | 1 | 2.1018052308895397E-62 1.0 |

| -0.6212406 -2.2146999 | 1 | 6.772270268679667E-97 1.0 |

| 11.1249309 9.9550664 | 0 | 1.0 3.1567838012477056E-56 |

| 9.9838097 10.9438362 | 0 | 1.0 1.9024447346702016E-51 |

| 10.8212212 10.5939013 | 0 | 1.0 2.800973098729602E-56 |

| 10.9189774 10.7821363 | 0 | 1.0 1.7209132744891742E-57 |

| 10.0745650 8.0106483 | 0 | 1.0 2.8642696635133805E-43 |

| 10.6198257 9.9438713 | 0 | 1.0 5.773273991940741E-53 |

| 9.8442045 8.5292476 | 0 | 1.0 2.527312305092764E-43 |

| 9.5218499 10.4179416 | 0 | 1.0 1.7314580596765114E-46 |