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

功能介绍

KMeans 是一个经典的聚类算法。基本思想是：以空间中k个点为中心进行聚类，对最靠近他们的对象归类。通过迭代的方法，逐次更新各聚类中心的值，直至得到最好的聚类结果。
本组件主要针对经纬度距离做Kmeans聚类，包括经纬度KMeans，经纬度KMeans预测, 经纬度KMeans流式预测。

经纬度距离（https://en.wikipedia.org/wiki/Haversine_formula)

经纬度K均值聚类预测 (GeoKMeansPredictStreamOp) - 图1经纬度K均值聚类预测 (GeoKMeansPredictStreamOp) - 图2输入数据中的经度和纬度使用度数表示，得到的距离单位为千米(km)。

参数说明

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

| 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 *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. df = pd.DataFrame([
  8.     [0, 0],
  9.     [8, 8],
  10.     [1, 2],
  11.     [9, 10],
  12.     [3, 1],
  13.     [10, 7]
  14. ])
  16. inOp1 = BatchOperator.fromDataframe(df, schemaStr='f0 long, f1 long')
  17. inOp2 = StreamOperator.fromDataframe(df, schemaStr='f0 long, f1 long')
  19. kmeans = GeoKMeansTrainBatchOp()\
  20.                 .setLongitudeCol("f0")\
  21.                 .setLatitudeCol("f1")\
  22.                 .setK(2)\
  23.                 .linkFrom(inOp1)
  24. kmeans.print()
  26. predict = GeoKMeansPredictBatchOp()\
  27.                 .setPredictionCol("pred")\
  28.                 .linkFrom(kmeans, inOp1)
  29. predict.print()
  31. predict = GeoKMeansPredictStreamOp(kmeans)\
  32.                 .setPredictionCol("pred")\
  33.                 .linkFrom(inOp2)
  34. predict.print()
  35. 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.KMeans4LongiLatitudePredictBatchOp;
  5. import com.alibaba.alink.operator.batch.clustering.KMeans4LongiLatitudeTrainBatchOp;
  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.KMeans4LongiLatitudePredictStreamOp;
  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 GeoKMeansPredictStreamOpTest {
  16.     @Test
  17.     public void testGeoKMeansPredictStreamOp() throws Exception {
  18.         List <Row> df = Arrays.asList(
  19.             Row.of(0, 0),
  20.             Row.of(8, 8),
  21.             Row.of(1, 2),
  22.             Row.of(9, 10),
  23.             Row.of(3, 1),
  24.             Row.of(10, 7)
  25.         );
  26.         BatchOperator <?> inOp1 = new MemSourceBatchOp(df, "f0 int, f1 int");
  27.         StreamOperator <?> inOp2 = new MemSourceStreamOp(df, "f0 int, f1 int");
  28.         BatchOperator <?> kmeans = new GeoKMeansTrainBatchOp()
  29.             .setLongitudeCol("f0")
  30.             .setLatitudeCol("f1")
  31.             .setK(2)
  32.             .linkFrom(inOp1);
  33.         kmeans.print();
  34.         BatchOperator <?> result = new GeoKMeansPredictBatchOp()
  35.             .setPredictionCol("pred")
  36.             .linkFrom(kmeans, inOp1);
  37.         result.print();
  38.         StreamOperator <?> predict = new GeoKMeansPredictStreamOp(kmeans)
  39.             .setPredictionCol("pred")
  40.             .linkFrom(inOp2);
  41.         predict.print();
  42.         StreamOperator.execute();
  43.     }
  44. }

运行结果

模型数据

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

| 0 | {“vectorCol”:null,”latitudeCol”:””f1””,”longitudeCol”:””f0””,”distanceType”:””HAVERSINE””,”k”:”2”,”vectorSize”:”2”} |

| 1048576 | {“clusterId”:0,”weight”:3.0,”center”:”[8.333333333333332, 9.0]”,”vec”:null} |

| 2097152 | {“clusterId”:1,”weight”:3.0,”center”:”[1.0, 1.3333333333333333]”,”vec”:null} |

预测输出

| f0 | f1 | pred | | —- | —- | —- |

| 0 | 0 | 1 |

| 8 | 8 | 0 |

| 1 | 2 | 1 |

| 9 | 10 | 0 |

| 3 | 1 | 1 |

| 10 | 7 | 0 |