Java 类名:com.alibaba.alink.operator.stream.clustering.KMeansPredictStreamOp
Python 类名:KMeansPredictStreamOp

功能介绍

KMeans 是一个经典的聚类算法。
基本思想是:以空间中k个点为中心进行聚类,对最靠近他们的对象归类。通过迭代的方法,逐次更新各聚类中心的值,直至得到最好的聚类结果。
Alink上KMeans算法包括KMeans,KMeans批量预测, KMeans流式预测。

参数说明

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

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

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

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

| predictionDistanceCol | 预测距离列名 | 预测距离列名 | 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 = pd.DataFrame([
  5. [0, "0 0 0"],
  6. [1, "0.1,0.1,0.1"],
  7. [2, "0.2,0.2,0.2"],
  8. [3, "9 9 9"],
  9. [4, "9.1 9.1 9.1"],
  10. [5, "9.2 9.2 9.2"]
  11. ])
  12. inOp1 = BatchOperator.fromDataframe(df, schemaStr='id int, vec string')
  13. inOp2 = StreamOperator.fromDataframe(df, schemaStr='id int, vec string')
  14. kmeans = KMeansTrainBatchOp()\
  15. .setVectorCol("vec")\
  16. .setK(2)\
  17. .linkFrom(inOp1)
  18. kmeans.lazyPrint(10)
  19. predictBatch = KMeansPredictBatchOp()\
  20. .setPredictionCol("pred")\
  21. .linkFrom(kmeans, inOp1)
  22. predictBatch.print()
  23. predictStream = KMeansPredictStreamOp(kmeans)\
  24. .setPredictionCol("pred")\
  25. .linkFrom(inOp2)
  26. predictStream.print()
  27. 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.KMeansPredictBatchOp;
  4. import com.alibaba.alink.operator.batch.clustering.KMeansTrainBatchOp;
  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.KMeansPredictStreamOp;
  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 KMeansPredictBatchOpTest {
  13. @Test
  14. public void testKMeansPredictBatchOp() throws Exception {
  15. List <Row> df = Arrays.asList(
  16. Row.of(0, "0 0 0"),
  17. Row.of(1, "0.1,0.1,0.1"),
  18. Row.of(2, "0.2,0.2,0.2"),
  19. Row.of(3, "9 9 9"),
  20. Row.of(4, "9.1 9.1 9.1"),
  21. Row.of(5, "9.2 9.2 9.2")
  22. );
  23. BatchOperator <?> inOp1 = new MemSourceBatchOp(df, "id int, vec string");
  24. StreamOperator <?> inOp2 = new MemSourceStreamOp(df, "id int, vec string");
  25. BatchOperator <?> kmeans = new KMeansTrainBatchOp()
  26. .setVectorCol("vec")
  27. .setK(2)
  28. .linkFrom(inOp1);
  29. kmeans.lazyPrint(10);
  30. BatchOperator <?> predictBatch = new KMeansPredictBatchOp()
  31. .setPredictionCol("pred")
  32. .linkFrom(kmeans, inOp1);
  33. predictBatch.print();
  34. StreamOperator <?> predictStream = new KMeansPredictStreamOp(kmeans)
  35. .setPredictionCol("pred")
  36. .linkFrom(inOp2);
  37. predictStream.print();
  38. StreamOperator.execute();
  39. }
  40. }

运行结果

模型结果

| model_id | model_info | | —- | —- |

| 0 | {“vectorCol”:””vec””,”latitudeCol”:null,”longitudeCol”:null,”distanceType”:””EUCLIDEAN””,”k”:”2”,”vectorSize”:”3”} |

| 1048576 | {“clusterId”:0,”weight”:3.0,”vec”:{“data”:[9.099999999999998,9.099999999999998,9.099999999999998]}} |

| 2097152 | {“clusterId”:1,”weight”:3.0,”vec”:{“data”:[0.1,0.1,0.1]}} |

预测结果

| id | vec | pred | | —- | —- | —- |

| 0 | 0 0 0 | 1 |

| 1 | 0.1,0.1,0.1 | 1 |

| 2 | 0.2,0.2,0.2 | 1 |

| 3 | 9 9 9 | 0 |

| 4 | 9.1 9.1 9.1 | 0 |

| 5 | 9.2 9.2 9.2 | 0 |