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

Java 代码

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

运行结果

| 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 |