You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/08/26 10:14:19 UTC
[GitHub] [flink-ml] zhipeng93 opened a new pull request, #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
zhipeng93 opened a new pull request, #148:
URL: https://github.com/apache/flink-ml/pull/148
## What is the purpose of the change
- Add AlgoOperator for AgglomerativeClustering[1] in Flink ML.
## Brief change log
- Added AlgoOperator for AgglomerativeClustering.
- Added java/python test/example for Transformer and Estimator for KBinsDiscretizer.
- Comparing with Sklearn [1], we made the following changes:
- The distance between two data points in sklearn could be `precomputed`, but we do not support this option in this PR
- sklearn uses `memory` to cache the output of the computation of the tree`, but we do not support this in this PR.
- sklearn uses `connectivity` to define for each sample the neighboring samples, but we do not support this in this PR.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (Java doc)
[1] https://scikit-learn.org/stable/modules/generated/sklearn.cluster.AgglomerativeClustering.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960328621
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
Good catch. I made a mistake --- and the mergeInfo should be different in different runs and only the distances should be the same accross different runs.
I have added a `map` after `env.fromCollection()` to force it as a parallel stream and also re-wrote the corresponding unit test.
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
Good catch. I made a mistake here --- and the mergeInfo should be different in different runs and only the distances should be the same accross different runs.
I have added a `map` after `env.fromCollection()` to force it as a parallel stream and also re-wrote the corresponding unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
lindong28 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961382143
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>The output contains two tables. The first one assigns one cluster Id for each data point. The
+ * second one contains the information of merging two clusters at each step. The data format of the
+ * merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer numCluster = getNumClusters();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (numCluster == null && distanceThreshold != null)
+ || (numCluster != null && distanceThreshold == null),
+ "One of param numCluster and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getNumClusters(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ mergeInfo.getTransformation().setParallelism(1);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer numCluster;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer numCluster,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.numCluster = numCluster;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ List<Row> inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (numCluster != null && activeClusters.size() > numCluster)
+ || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
+ (numCluster != null && activeClusters.size() > numCluster)
+ || (distanceThreshold != null && distanceThreshold > minDistance);
+ }
+ }
+
+ private double computeDistanceBetweenClusters(Cluster cluster1, Cluster cluster2) {
+ double distance;
+ int size1 = cluster1.dataPointIds.size();
+ int size2 = cluster2.dataPointIds.size();
+
+ switch (linkage) {
+ case LINKAGE_AVERAGE:
+ distance = 0;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance += distanceMeasure.distance(vectorWithNorm1, vectorWithNorm2);
+ }
+ }
+ distance /= size1 * size2;
+ break;
+ case LINKAGE_COMPLETE:
+ distance = Double.MIN_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.max(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_SINGLE:
+ distance = Double.MAX_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.min(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_WARD:
+ int vecSize = vectorWithNorms[0].vector.size();
+ DenseVector mean1 = Vectors.dense(new double[vecSize]);
+ DenseVector mean2 = Vectors.dense(new double[vecSize]);
+
+ for (int i = 0; i < size1; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster1.dataPointIds.get(i)].vector, mean1);
+ }
+ for (int i = 0; i < size2; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster2.dataPointIds.get(i)].vector, mean2);
+ }
+
+ DenseVector meanMerged = mean1.clone();
+ BLAS.axpy(1.0, mean2, meanMerged);
+ BLAS.scal(1.0 / size1, mean1);
+ BLAS.scal(1.0 / size2, mean2);
+ BLAS.scal(1.0 / (size1 + size2), meanMerged);
+ double essInc =
+ size1 * BLAS.dot(mean1, mean1)
+ + size2 * BLAS.dot(mean2, mean2)
+ - (size1 + size2) * BLAS.dot(meanMerged, meanMerged);
+
+ distance = Math.sqrt(2 * essInc);
Review Comment:
Since the calculated distance is used to find the pair of clusters that minimize this distance, would it be more efficient to just return `essInc` as the distance?
Suppose `distance_1 = Math.sqrt(2 * essInc_1)` and `distance_2 = Math.sqrt(2 * essInc_2)`, it is guaranteed that `distance_1 < distance_2` if and only if `essInc_1 < essInc_2`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960328621
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
Good catch. I made a mistake here --- the mergeInfo should be different in different runs and only the distances should be the same accross different runs.
I have added a `map` after `env.fromCollection()` to force it as a parallel stream and also re-wrote the corresponding unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961545453
##########
flink-ml-python/pyflink/ml/lib/clustering/agglomerativeclustering.py:
##########
@@ -0,0 +1,135 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, \
+ BooleanParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.clustering.common import JavaClusteringAlgoOperator
+from pyflink.ml.lib.param import HasDistanceMeasure, HasFeaturesCol, HasPredictionCol
+
+
+class _AgglomerativeClusteringParams(
+ JavaWithParams,
+ HasDistanceMeasure,
+ HasFeaturesCol,
+ HasPredictionCol
+):
+ """
+ Params for :class:`AgglomerativeClustering`.
+ """
+ NUM_CLUSTERS: Param[int] = IntParam("num_clusters",
+ "The max number of clusters to create.",
+ 2)
+
+ DISTANCE_THRESHOLD: Param[float] = \
+ FloatParam("distance_threshold",
+ "Threshold to decide whether two clusters should be merged.",
+ None)
+
+ """
+ Supported options to compute the distance between two clusters.
+ <ul>
+ <li>ward: difference between the sum of the variance of the two clusters and the merged one.
+ <li>complete: the maximum distance between all observations of the two clusters.
+ <li>single: the minimum distance between all observations of the two cluster.
+ <li>average: the average of the distance of all observations of the two cluster.
+ </ul>
+ """
+ LINKAGE: Param[str] = StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ "ward",
+ ParamValidators.in_array(
+ ["ward", "complete", "single", "average"]))
+
+ COMPUTE_FULL_TREE: Param[bool] = BooleanParam(
+ "compute_full_tree",
+ "Whether computes the full tree after convergence.",
+ False,
+ ParamValidators.not_null())
+
+ def __init__(self, java_params):
+ super(_AgglomerativeClusteringParams, self).__init__(java_params)
+
+ def set_num_clusters(self, value: int):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.NUM_CLUSTERS, value))
+
+ def get_num_clusters(self) -> int:
+ return self.get(self.NUM_CLUSTERS)
+
+ def set_distance_threshold(self, value: float):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.DISTANCE_THRESHOLD, value))
+
+ def get_distance_threshold(self) -> float:
+ return self.get(self.DISTANCE_THRESHOLD)
+
+ def set_linkage(self, value: str):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.LINKAGE, value))
+
+ def get_linkage(self) -> str:
+ return self.get(self.LINKAGE)
+
+ def set_compute_full_tree(self, value: bool):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.COMPUTE_FULL_TREE, value))
+
+ def get_compute_full_tree(self) -> bool:
+ return self.get(self.COMPUTE_FULL_TREE)
+
+ @property
+ def num_clusters(self):
+ return self.get_num_clusters()
+
+ @property
+ def distance_threshold(self):
+ return self.get_distance_threshold()
+
+ @property
+ def linkage(self):
+ return self.get_linkage()
+
+ @property
+ def compute_full_tree(self):
+ return self.get_compute_full_tree()
+
+
+class AgglomerativeClustering(JavaClusteringAlgoOperator, _AgglomerativeClusteringParams):
+ """
+ An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ observation starts in its own cluster and the clusters are merged together one by one.
+ Users can choose different strategies to merge two clusters by setting
+ {@link AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting
+ {@link AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+
+ <p>The output contains two tables. The first one assigns one cluster Id for each data point.
+ The second one contains the information of merging two clusters at each step. The data format
+ of the merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
Review Comment:
Thanks for the comment. The script to visualize the merge info is added in the python example script[1] and the user doc[2].
[1] flink-ml-python/pyflink/examples/ml/clustering/agglomerativeclustering_example.py
[2] docs/content/docs/operators/clustering/agglomerativeclustering.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 merged pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
lindong28 merged PR #148:
URL: https://github.com/apache/flink-ml/pull/148
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961513789
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>The output contains two tables. The first one assigns one cluster Id for each data point. The
+ * second one contains the information of merging two clusters at each step. The data format of the
+ * merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer numCluster = getNumClusters();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (numCluster == null && distanceThreshold != null)
+ || (numCluster != null && distanceThreshold == null),
+ "One of param numCluster and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getNumClusters(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ mergeInfo.getTransformation().setParallelism(1);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer numCluster;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer numCluster,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.numCluster = numCluster;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ List<Row> inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (numCluster != null && activeClusters.size() > numCluster)
+ || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
+ (numCluster != null && activeClusters.size() > numCluster)
+ || (distanceThreshold != null && distanceThreshold > minDistance);
+ }
+ }
+
+ private double computeDistanceBetweenClusters(Cluster cluster1, Cluster cluster2) {
+ double distance;
+ int size1 = cluster1.dataPointIds.size();
+ int size2 = cluster2.dataPointIds.size();
+
+ switch (linkage) {
+ case LINKAGE_AVERAGE:
+ distance = 0;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance += distanceMeasure.distance(vectorWithNorm1, vectorWithNorm2);
+ }
+ }
+ distance /= size1 * size2;
+ break;
+ case LINKAGE_COMPLETE:
+ distance = Double.MIN_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.max(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_SINGLE:
+ distance = Double.MAX_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.min(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_WARD:
+ int vecSize = vectorWithNorms[0].vector.size();
+ DenseVector mean1 = Vectors.dense(new double[vecSize]);
+ DenseVector mean2 = Vectors.dense(new double[vecSize]);
+
+ for (int i = 0; i < size1; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster1.dataPointIds.get(i)].vector, mean1);
+ }
+ for (int i = 0; i < size2; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster2.dataPointIds.get(i)].vector, mean2);
+ }
+
+ DenseVector meanMerged = mean1.clone();
+ BLAS.axpy(1.0, mean2, meanMerged);
+ BLAS.scal(1.0 / size1, mean1);
+ BLAS.scal(1.0 / size2, mean2);
+ BLAS.scal(1.0 / (size1 + size2), meanMerged);
+ double essInc =
+ size1 * BLAS.dot(mean1, mean1)
+ + size2 * BLAS.dot(mean2, mean2)
+ - (size1 + size2) * BLAS.dot(meanMerged, meanMerged);
+
+ distance = Math.sqrt(2 * essInc);
Review Comment:
The distance will be returned as merge info, which is consistent with sklearn's implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on PR #148:
URL: https://github.com/apache/flink-ml/pull/148#issuecomment-1230125226
@lindong28 @yunfengzhou-hub Can you help to review this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960328621
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
Good catch. I made a mistake here --- the mergeInfo should be different in different runs and only the distances should be the same.
I have added a `map` after `env.fromCollection()` to force it as a parallel stream and also re-wrote the corresponding unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960175355
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> K = new IntParam("k", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
+ * one.
+ * <li>complete: the maximum distance between all observations of the two clusters.
+ * <li>single: the minimum distance between all observations of the two cluster.
+ * <li>average: the average of the distance of all observations of the two cluster.
+ * </ul>
+ */
+ Param<String> LINKAGE =
+ new StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ LINKAGE_WARD,
+ ParamValidators.inArray(
+ LINKAGE_WARD, LINKAGE_COMPLETE, LINKAGE_AVERAGE, LINKAGE_SINGLE));
+
+ Param<Boolean> COMPUTE_FULL_TREE =
+ new BooleanParam(
+ "computeFullTree",
+ "Whether computes the full tree after convergence.",
+ false,
Review Comment:
With some offline discussion, we aggree that not supporting `auto` is ok for users.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r958215741
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer k,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.k = k;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (k != null && activeClusters.size() > k) || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
Review Comment:
No, we need to merge the cluster in this round first.
Say when entering the loop, the number of active clusters is `K+1`. If we change the order as above, the number of output clusters will be `k+1`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960328621
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
Good catch. I made a mistake here --- the mergeInfo should be different in different runs and only the distances should be the same.
I have added a `map` after `env.fromCollection()` to force it to be a parallel stream and also re-wrote the corresponding unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960296835
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
Review Comment:
Thanks for the explanation. I agree with it that the current API is better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
lindong28 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961352889
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
Review Comment:
It seems inconsistent to have `merged one` for this option but not the options described below. And after reading sklearn's doc for this parameter and the source code, I am not sure it is intuitive to mention `sum` here.
How about this:
`ward: variance between the two clusters`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
lindong28 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961348986
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
+ * one.
+ * <li>complete: the maximum distance between all observations of the two clusters.
+ * <li>single: the minimum distance between all observations of the two cluster.
+ * <li>average: the average of the distance of all observations of the two cluster.
+ * </ul>
+ */
+ Param<String> LINKAGE =
+ new StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
Review Comment:
Would it be useful to mention `The algorithm will merge the pairs of cluster that minimize this criterion` similar to sklearn's doc?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
+ * one.
+ * <li>complete: the maximum distance between all observations of the two clusters.
+ * <li>single: the minimum distance between all observations of the two cluster.
+ * <li>average: the average of the distance of all observations of the two cluster.
Review Comment:
Would the following be more consistent with the options described above:
```
the average distance between all observations of the two clusters.
```
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
+ * one.
+ * <li>complete: the maximum distance between all observations of the two clusters.
+ * <li>single: the minimum distance between all observations of the two cluster.
Review Comment:
cluster -> clusters
##########
flink-ml-python/pyflink/ml/lib/clustering/agglomerativeclustering.py:
##########
@@ -0,0 +1,135 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, \
+ BooleanParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.clustering.common import JavaClusteringAlgoOperator
+from pyflink.ml.lib.param import HasDistanceMeasure, HasFeaturesCol, HasPredictionCol
+
+
+class _AgglomerativeClusteringParams(
+ JavaWithParams,
+ HasDistanceMeasure,
+ HasFeaturesCol,
+ HasPredictionCol
+):
+ """
+ Params for :class:`AgglomerativeClustering`.
+ """
+ NUM_CLUSTERS: Param[int] = IntParam("num_clusters",
+ "The max number of clusters to create.",
+ 2)
+
+ DISTANCE_THRESHOLD: Param[float] = \
+ FloatParam("distance_threshold",
+ "Threshold to decide whether two clusters should be merged.",
+ None)
+
+ """
+ Supported options to compute the distance between two clusters.
+ <ul>
+ <li>ward: difference between the sum of the variance of the two clusters and the merged one.
+ <li>complete: the maximum distance between all observations of the two clusters.
+ <li>single: the minimum distance between all observations of the two cluster.
+ <li>average: the average of the distance of all observations of the two cluster.
+ </ul>
+ """
+ LINKAGE: Param[str] = StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ "ward",
+ ParamValidators.in_array(
+ ["ward", "complete", "single", "average"]))
+
+ COMPUTE_FULL_TREE: Param[bool] = BooleanParam(
+ "compute_full_tree",
+ "Whether computes the full tree after convergence.",
+ False,
+ ParamValidators.not_null())
+
+ def __init__(self, java_params):
+ super(_AgglomerativeClusteringParams, self).__init__(java_params)
+
+ def set_num_clusters(self, value: int):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.NUM_CLUSTERS, value))
+
+ def get_num_clusters(self) -> int:
+ return self.get(self.NUM_CLUSTERS)
+
+ def set_distance_threshold(self, value: float):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.DISTANCE_THRESHOLD, value))
+
+ def get_distance_threshold(self) -> float:
+ return self.get(self.DISTANCE_THRESHOLD)
+
+ def set_linkage(self, value: str):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.LINKAGE, value))
+
+ def get_linkage(self) -> str:
+ return self.get(self.LINKAGE)
+
+ def set_compute_full_tree(self, value: bool):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.COMPUTE_FULL_TREE, value))
+
+ def get_compute_full_tree(self) -> bool:
+ return self.get(self.COMPUTE_FULL_TREE)
+
+ @property
+ def num_clusters(self):
+ return self.get_num_clusters()
+
+ @property
+ def distance_threshold(self):
+ return self.get_distance_threshold()
+
+ @property
+ def linkage(self):
+ return self.get_linkage()
+
+ @property
+ def compute_full_tree(self):
+ return self.get_compute_full_tree()
+
+
+class AgglomerativeClustering(JavaClusteringAlgoOperator, _AgglomerativeClusteringParams):
+ """
+ An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
Review Comment:
bottom up -> bottom-up
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
Review Comment:
It seems inconsistent to have `merged one` for this option but not the options described below.
How about this:
`ward: the sum of variance of the two clusters`
##########
flink-ml-python/pyflink/ml/lib/clustering/agglomerativeclustering.py:
##########
@@ -0,0 +1,135 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, \
+ BooleanParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.clustering.common import JavaClusteringAlgoOperator
+from pyflink.ml.lib.param import HasDistanceMeasure, HasFeaturesCol, HasPredictionCol
+
+
+class _AgglomerativeClusteringParams(
+ JavaWithParams,
+ HasDistanceMeasure,
+ HasFeaturesCol,
+ HasPredictionCol
+):
+ """
+ Params for :class:`AgglomerativeClustering`.
+ """
+ NUM_CLUSTERS: Param[int] = IntParam("num_clusters",
+ "The max number of clusters to create.",
+ 2)
+
+ DISTANCE_THRESHOLD: Param[float] = \
+ FloatParam("distance_threshold",
+ "Threshold to decide whether two clusters should be merged.",
+ None)
+
+ """
+ Supported options to compute the distance between two clusters.
+ <ul>
+ <li>ward: difference between the sum of the variance of the two clusters and the merged one.
+ <li>complete: the maximum distance between all observations of the two clusters.
+ <li>single: the minimum distance between all observations of the two cluster.
+ <li>average: the average of the distance of all observations of the two cluster.
+ </ul>
+ """
+ LINKAGE: Param[str] = StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ "ward",
+ ParamValidators.in_array(
+ ["ward", "complete", "single", "average"]))
+
+ COMPUTE_FULL_TREE: Param[bool] = BooleanParam(
+ "compute_full_tree",
+ "Whether computes the full tree after convergence.",
+ False,
+ ParamValidators.not_null())
+
+ def __init__(self, java_params):
+ super(_AgglomerativeClusteringParams, self).__init__(java_params)
+
+ def set_num_clusters(self, value: int):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.NUM_CLUSTERS, value))
+
+ def get_num_clusters(self) -> int:
+ return self.get(self.NUM_CLUSTERS)
+
+ def set_distance_threshold(self, value: float):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.DISTANCE_THRESHOLD, value))
+
+ def get_distance_threshold(self) -> float:
+ return self.get(self.DISTANCE_THRESHOLD)
+
+ def set_linkage(self, value: str):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.LINKAGE, value))
+
+ def get_linkage(self) -> str:
+ return self.get(self.LINKAGE)
+
+ def set_compute_full_tree(self, value: bool):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.COMPUTE_FULL_TREE, value))
+
+ def get_compute_full_tree(self) -> bool:
+ return self.get(self.COMPUTE_FULL_TREE)
+
+ @property
+ def num_clusters(self):
+ return self.get_num_clusters()
+
+ @property
+ def distance_threshold(self):
+ return self.get_distance_threshold()
+
+ @property
+ def linkage(self):
+ return self.get_linkage()
+
+ @property
+ def compute_full_tree(self):
+ return self.get_compute_full_tree()
+
+
+class AgglomerativeClustering(JavaClusteringAlgoOperator, _AgglomerativeClusteringParams):
+ """
+ An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ observation starts in its own cluster and the clusters are merged together one by one.
+ Users can choose different strategies to merge two clusters by setting
+ {@link AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting
+ {@link AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+
+ <p>The output contains two tables. The first one assigns one cluster Id for each data point.
+ The second one contains the information of merging two clusters at each step. The data format
+ of the merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
Review Comment:
Maybe provide path to the scripts that visualize the 2nd output.
##########
flink-ml-python/pyflink/ml/lib/clustering/agglomerativeclustering.py:
##########
@@ -0,0 +1,135 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, \
+ BooleanParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.clustering.common import JavaClusteringAlgoOperator
+from pyflink.ml.lib.param import HasDistanceMeasure, HasFeaturesCol, HasPredictionCol
+
+
+class _AgglomerativeClusteringParams(
+ JavaWithParams,
+ HasDistanceMeasure,
+ HasFeaturesCol,
+ HasPredictionCol
+):
+ """
+ Params for :class:`AgglomerativeClustering`.
+ """
+ NUM_CLUSTERS: Param[int] = IntParam("num_clusters",
+ "The max number of clusters to create.",
+ 2)
+
+ DISTANCE_THRESHOLD: Param[float] = \
+ FloatParam("distance_threshold",
+ "Threshold to decide whether two clusters should be merged.",
+ None)
+
+ """
+ Supported options to compute the distance between two clusters.
+ <ul>
+ <li>ward: difference between the sum of the variance of the two clusters and the merged one.
+ <li>complete: the maximum distance between all observations of the two clusters.
+ <li>single: the minimum distance between all observations of the two cluster.
+ <li>average: the average of the distance of all observations of the two cluster.
+ </ul>
+ """
+ LINKAGE: Param[str] = StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ "ward",
+ ParamValidators.in_array(
+ ["ward", "complete", "single", "average"]))
+
+ COMPUTE_FULL_TREE: Param[bool] = BooleanParam(
+ "compute_full_tree",
+ "Whether computes the full tree after convergence.",
+ False,
+ ParamValidators.not_null())
+
+ def __init__(self, java_params):
+ super(_AgglomerativeClusteringParams, self).__init__(java_params)
+
+ def set_num_clusters(self, value: int):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.NUM_CLUSTERS, value))
+
+ def get_num_clusters(self) -> int:
+ return self.get(self.NUM_CLUSTERS)
+
+ def set_distance_threshold(self, value: float):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.DISTANCE_THRESHOLD, value))
+
+ def get_distance_threshold(self) -> float:
+ return self.get(self.DISTANCE_THRESHOLD)
+
+ def set_linkage(self, value: str):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.LINKAGE, value))
+
+ def get_linkage(self) -> str:
+ return self.get(self.LINKAGE)
+
+ def set_compute_full_tree(self, value: bool):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.COMPUTE_FULL_TREE, value))
+
+ def get_compute_full_tree(self) -> bool:
+ return self.get(self.COMPUTE_FULL_TREE)
+
+ @property
+ def num_clusters(self):
+ return self.get_num_clusters()
+
+ @property
+ def distance_threshold(self):
+ return self.get_distance_threshold()
+
+ @property
+ def linkage(self):
+ return self.get_linkage()
+
+ @property
+ def compute_full_tree(self):
+ return self.get_compute_full_tree()
+
+
+class AgglomerativeClustering(JavaClusteringAlgoOperator, _AgglomerativeClusteringParams):
+ """
+ An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ observation starts in its own cluster and the clusters are merged together one by one.
+ Users can choose different strategies to merge two clusters by setting
+ {@link AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting
Review Comment:
distance measure -> distance measures
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r958253423
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
Review Comment:
Thanks for the comment. I have added the following description in Java doc.
```
The output contains two tables. The first one assigns one cluster Id for each data point.
The second one contains the information of merging two clusters at each step. The data format
of the merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
```
I am not sure about adding a `Table getMergeInfo()` here, since it is somehow adding a public API.
For the semantic of `AgglomerativeClustering`, I made it an AlgoOperator because (1) it is not a Estimator since it has not model data and cannot be used to predict some new data. (2) It is also not a Transformer.
It is clear that not all AlgoOperator/Estimator could be used in Pipeline. For those multi-input/output stages, we have `Graph`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960171818
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
Review Comment:
If the input data comes from an operator whose parallelism is 3, how can we make sure that they will arrive at the `LocalAgglomerativeClusteringOperator`, whose parallelism is 1, in the same order and be assigned the same cluster ids across difference runs? If each record might be assigned a different id in different runs, why would the merge info remains the same and can be verified with an expected value?
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
+ }
+
+ @Test
+ public void testParam() {
+ AgglomerativeClustering agglomerativeClustering = new AgglomerativeClustering();
+ assertEquals("features", agglomerativeClustering.getFeaturesCol());
+ assertEquals(2, agglomerativeClustering.getK().intValue());
+ assertNull(agglomerativeClustering.getDistanceThreshold());
+ assertEquals(AgglomerativeClustering.LINKAGE_WARD, agglomerativeClustering.getLinkage());
+ assertEquals(EuclideanDistanceMeasure.NAME, agglomerativeClustering.getDistanceMeasure());
+ assertFalse(agglomerativeClustering.getComputeFullTree());
+ assertEquals("prediction", agglomerativeClustering.getPredictionCol());
+
+ agglomerativeClustering
+ .setFeaturesCol("test_features")
+ .setK(null)
+ .setDistanceThreshold(0.01)
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(CosineDistanceMeasure.NAME)
+ .setComputeFullTree(true)
+ .setPredictionCol("cluster_id");
+
+ assertEquals("test_features", agglomerativeClustering.getFeaturesCol());
+ assertNull(agglomerativeClustering.getK());
+ assertEquals(0.01, agglomerativeClustering.getDistanceThreshold(), TOLERANCE);
+ assertEquals(AgglomerativeClustering.LINKAGE_AVERAGE, agglomerativeClustering.getLinkage());
+ assertEquals(CosineDistanceMeasure.NAME, agglomerativeClustering.getDistanceMeasure());
+ assertTrue(agglomerativeClustering.getComputeFullTree());
+ assertEquals("cluster_id", agglomerativeClustering.getPredictionCol());
+ }
+
+ @Test
+ public void testOutputSchema() {
+ Table tempTable =
+ tEnv.fromDataStream(env.fromElements(Row.of("", "")))
+ .as("test_features", "dummy_input");
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setFeaturesCol("test_features")
+ .setPredictionCol("test_prediction");
+ Table[] outputs = agglomerativeClustering.transform(tempTable);
+ assertEquals(2, outputs.length);
+ assertEquals(
+ Arrays.asList("test_features", "dummy_input", "test_prediction"),
+ outputs[0].getResolvedSchema().getColumnNames());
+ assertEquals(
+ Arrays.asList("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster"),
+ outputs[1].getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testTransform() throws Exception {
+ Table[] outputs;
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred");
+
+ // Tests euclidean distance with linkage as average, k = 2.
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+
+ // Tests euclidean distance with linkage as average, k = 2, compute_full_tree = true.
+ outputs = agglomerativeClustering.setComputeFullTree(true).transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+
+ // Tests euclidean distance with linkage as average, distance_threshold = 2.
+ outputs =
+ agglomerativeClustering
+ .setK(null)
+ .setDistanceThreshold(2.0)
+ .transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT, outputs[0]);
+ }
+
+ @Test
+ public void testMergeInfo() throws Exception {
+ Table[] outputs;
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred")
+ .setComputeFullTree(true);
+
+ // Tests euclidean distance with linkage as average.
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests cosine distance with linkage as average.
+ agglomerativeClustering.setDistanceMeasure(CosineDistanceMeasure.NAME);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(COSINE_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests manhattan distance with linkage as average.
+ agglomerativeClustering.setDistanceMeasure(ManhattanDistanceMeasure.NAME);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(MANHATTAN_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as complete.
+ agglomerativeClustering
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_COMPLETE);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_COMPLETE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as single.
+ agglomerativeClustering.setLinkage(AgglomerativeClusteringParams.LINKAGE_SINGLE);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_SINGLE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as ward.
+ agglomerativeClustering.setLinkage(AgglomerativeClusteringParams.LINKAGE_WARD);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_WARD_MERGE_INFO, outputs[1]);
+
+ // Tests merge info not fully computed.
+ agglomerativeClustering.setComputeFullTree(false);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_WARD_MERGE_INFO.subList(0, 4), outputs[1]);
+ }
+
+ @Test
+ public void testSaveLoadTransform() throws Exception {
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred");
+
+ agglomerativeClustering =
+ TestUtils.saveAndReload(
+ tEnv, agglomerativeClustering, tempFolder.newFolder().getAbsolutePath());
+
+ Table[] outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void verifyMergeInfo(List<Row> expected, Table mergeInfoTable) throws Exception {
+ List<Row> mergeInfo =
+ IteratorUtils.toList(tEnv.toDataStream(mergeInfoTable).executeAndCollect());
+ assertEquals(expected.size(), mergeInfo.size());
+ mergeInfo.sort(Comparator.comparingInt(o -> o.getFieldAs(0)));
Review Comment:
`mergeInfo` should be order-sensitive. I think we should verify that the order of the output merge information is also correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
lindong28 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961557819
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> NUM_CLUSTERS =
+ new IntParam("numClusters", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters. The algorithm will merge the
+ * pairs of cluster that minimize this criterion
Review Comment:
nits: Add `.` at the end?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r960325335
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/AgglomerativeClusteringTest.java:
##########
@@ -0,0 +1,312 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClustering;
+import org.apache.flink.ml.clustering.agglomerativeclustering.AgglomerativeClusteringParams;
+import org.apache.flink.ml.common.distance.CosineDistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.common.distance.ManhattanDistanceMeasure;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** Tests {@link AgglomerativeClustering}. */
+public class AgglomerativeClusteringTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamTableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private Table inputDataTable;
+
+ private static final List<DenseVector> INPUT_DATA =
+ Arrays.asList(
+ Vectors.dense(1, 1),
+ Vectors.dense(1, 4),
+ Vectors.dense(1, 0),
+ Vectors.dense(4, 1.5),
+ Vectors.dense(4, 4),
+ Vectors.dense(4, 0));
+
+ private static final List<Row> EUCLIDEAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.1394402, 4),
+ Row.of(8, 9, 3.9559706, 6));
+
+ private static final List<Row> COSINE_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(2, 5, 0, 2),
+ Row.of(0, 4, 1.1102230E-16, 2),
+ Row.of(3, 6, 0.0636708, 3),
+ Row.of(1, 7, 0.1425070, 3),
+ Row.of(8, 9, 0.3664484, 6));
+
+ private static final List<Row> MANHATTAN_AVERAGE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.75, 4),
+ Row.of(8, 9, 4.875, 6));
+
+ private static final List<Row> EUCLIDEAN_SINGLE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(4, 7, 2.5, 3),
+ Row.of(8, 9, 3, 6),
+ Row.of(1, 6, 3, 3));
+
+ private static final List<Row> EUCLIDEAN_WARD_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 4.2573465, 4),
+ Row.of(8, 9, 5.5113519, 6));
+
+ private static final List<Row> EUCLIDEAN_COMPLETE_MERGE_INFO =
+ Arrays.asList(
+ Row.of(0, 2, 1, 2),
+ Row.of(3, 5, 1.5, 2),
+ Row.of(1, 4, 3, 2),
+ Row.of(6, 7, 3.3541019, 4),
+ Row.of(8, 9, 5, 6));
+
+ private static final List<Row> EUCLIDEAN_WARD_K_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 0),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 0));
+
+ private static final List<Row> EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT =
+ Arrays.asList(
+ Row.of(Vectors.dense(1, 1), 0),
+ Row.of(Vectors.dense(1, 4), 1),
+ Row.of(Vectors.dense(1, 0), 0),
+ Row.of(Vectors.dense(4, 1.5), 2),
+ Row.of(Vectors.dense(4, 4), 1),
+ Row.of(Vectors.dense(4, 0), 2));
+
+ private static final double TOLERANCE = 1e-7;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(3);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+ inputDataTable = tEnv.fromDataStream(env.fromCollection(INPUT_DATA)).as("features");
+ }
+
+ @Test
+ public void testParam() {
+ AgglomerativeClustering agglomerativeClustering = new AgglomerativeClustering();
+ assertEquals("features", agglomerativeClustering.getFeaturesCol());
+ assertEquals(2, agglomerativeClustering.getK().intValue());
+ assertNull(agglomerativeClustering.getDistanceThreshold());
+ assertEquals(AgglomerativeClustering.LINKAGE_WARD, agglomerativeClustering.getLinkage());
+ assertEquals(EuclideanDistanceMeasure.NAME, agglomerativeClustering.getDistanceMeasure());
+ assertFalse(agglomerativeClustering.getComputeFullTree());
+ assertEquals("prediction", agglomerativeClustering.getPredictionCol());
+
+ agglomerativeClustering
+ .setFeaturesCol("test_features")
+ .setK(null)
+ .setDistanceThreshold(0.01)
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(CosineDistanceMeasure.NAME)
+ .setComputeFullTree(true)
+ .setPredictionCol("cluster_id");
+
+ assertEquals("test_features", agglomerativeClustering.getFeaturesCol());
+ assertNull(agglomerativeClustering.getK());
+ assertEquals(0.01, agglomerativeClustering.getDistanceThreshold(), TOLERANCE);
+ assertEquals(AgglomerativeClustering.LINKAGE_AVERAGE, agglomerativeClustering.getLinkage());
+ assertEquals(CosineDistanceMeasure.NAME, agglomerativeClustering.getDistanceMeasure());
+ assertTrue(agglomerativeClustering.getComputeFullTree());
+ assertEquals("cluster_id", agglomerativeClustering.getPredictionCol());
+ }
+
+ @Test
+ public void testOutputSchema() {
+ Table tempTable =
+ tEnv.fromDataStream(env.fromElements(Row.of("", "")))
+ .as("test_features", "dummy_input");
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setFeaturesCol("test_features")
+ .setPredictionCol("test_prediction");
+ Table[] outputs = agglomerativeClustering.transform(tempTable);
+ assertEquals(2, outputs.length);
+ assertEquals(
+ Arrays.asList("test_features", "dummy_input", "test_prediction"),
+ outputs[0].getResolvedSchema().getColumnNames());
+ assertEquals(
+ Arrays.asList("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster"),
+ outputs[1].getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testTransform() throws Exception {
+ Table[] outputs;
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred");
+
+ // Tests euclidean distance with linkage as average, k = 2.
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+
+ // Tests euclidean distance with linkage as average, k = 2, compute_full_tree = true.
+ outputs = agglomerativeClustering.setComputeFullTree(true).transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+
+ // Tests euclidean distance with linkage as average, distance_threshold = 2.
+ outputs =
+ agglomerativeClustering
+ .setK(null)
+ .setDistanceThreshold(2.0)
+ .transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_THRESHOLD_AS_TWO_RESULT, outputs[0]);
+ }
+
+ @Test
+ public void testMergeInfo() throws Exception {
+ Table[] outputs;
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred")
+ .setComputeFullTree(true);
+
+ // Tests euclidean distance with linkage as average.
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests cosine distance with linkage as average.
+ agglomerativeClustering.setDistanceMeasure(CosineDistanceMeasure.NAME);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(COSINE_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests manhattan distance with linkage as average.
+ agglomerativeClustering.setDistanceMeasure(ManhattanDistanceMeasure.NAME);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(MANHATTAN_AVERAGE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as complete.
+ agglomerativeClustering
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_COMPLETE);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_COMPLETE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as single.
+ agglomerativeClustering.setLinkage(AgglomerativeClusteringParams.LINKAGE_SINGLE);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_SINGLE_MERGE_INFO, outputs[1]);
+
+ // Tests euclidean distance with linkage as ward.
+ agglomerativeClustering.setLinkage(AgglomerativeClusteringParams.LINKAGE_WARD);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_WARD_MERGE_INFO, outputs[1]);
+
+ // Tests merge info not fully computed.
+ agglomerativeClustering.setComputeFullTree(false);
+ outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyMergeInfo(EUCLIDEAN_WARD_MERGE_INFO.subList(0, 4), outputs[1]);
+ }
+
+ @Test
+ public void testSaveLoadTransform() throws Exception {
+ AgglomerativeClustering agglomerativeClustering =
+ new AgglomerativeClustering()
+ .setLinkage(AgglomerativeClusteringParams.LINKAGE_AVERAGE)
+ .setDistanceMeasure(EuclideanDistanceMeasure.NAME)
+ .setPredictionCol("pred");
+
+ agglomerativeClustering =
+ TestUtils.saveAndReload(
+ tEnv, agglomerativeClustering, tempFolder.newFolder().getAbsolutePath());
+
+ Table[] outputs = agglomerativeClustering.transform(inputDataTable);
+ verifyClusteringResult(EUCLIDEAN_WARD_K_AS_TWO_RESULT, outputs[0]);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void verifyMergeInfo(List<Row> expected, Table mergeInfoTable) throws Exception {
+ List<Row> mergeInfo =
+ IteratorUtils.toList(tEnv.toDataStream(mergeInfoTable).executeAndCollect());
+ assertEquals(expected.size(), mergeInfo.size());
+ mergeInfo.sort(Comparator.comparingInt(o -> o.getFieldAs(0)));
Review Comment:
Good catch. I have changed the unit test to verify the order of the merge information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] yunfengzhou-hub commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r957971688
##########
flink-ml-python/pyflink/ml/lib/tests/test_ml_lib_completeness.py:
##########
@@ -70,7 +70,7 @@ def _load_stages_from_module(cls, module):
'JavaClusteringEstimator', 'JavaClusteringModel',
'JavaEvaluationAlgoOperator', 'JavaFeatureTransformer',
'JavaFeatureEstimator', 'JavaFeatureModel',
- 'JavaRegressionEstimator', 'JavaRegressionModel')]
+ 'JavaRegressionEstimator', 'JavaRegressionModel', 'JavaClusteringAlgoOperator')]
Review Comment:
nit: it might be better to place `JavaClusteringAlgoOperator` between `JavaClassificationModel` and `JavaClusteringEstimator`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
Review Comment:
It seems that this variable can be converted to a local variable.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer k,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.k = k;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (k != null && activeClusters.size() > k) || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
Review Comment:
How about changing the order of the logics here?
```java
clusteringRunning =
(k != null && activeClusters.size() > k)
|| (distanceThreshold != null && distanceThreshold > minDistance);
if (clusteringRunning) {
int mergedClusterId = mergedCluster.clusterId;
for (int dataPointId : mergedCluster.dataPointIds) {
clusterIds[dataPointId] = mergedClusterId;
}
}
```
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer k,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.k = k;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (k != null && activeClusters.size() > k) || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
+ (k != null && activeClusters.size() > k)
+ || (distanceThreshold != null && distanceThreshold > minDistance);
+ }
+ }
+
+ private double computeDistanceBetweenClusters(Cluster cluster1, Cluster cluster2) {
+ double distance;
+ int size1 = cluster1.dataPointIds.size();
+ int size2 = cluster2.dataPointIds.size();
+
+ switch (linkage) {
+ case LINKAGE_AVERAGE:
+ distance = 0;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance += distanceMeasure.distance(vectorWithNorm1, vectorWithNorm2);
+ }
+ }
+ distance /= size1 * size2;
+ break;
+ case LINKAGE_COMPLETE:
+ distance = Double.MIN_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.max(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_SINGLE:
+ distance = Double.MAX_VALUE;
+ for (int i = 0; i < size1; i++) {
+ for (int j = 0; j < size2; j++) {
+ VectorWithNorm vectorWithNorm1 =
+ vectorWithNorms[cluster1.dataPointIds.get(i)];
+ VectorWithNorm vectorWithNorm2 =
+ vectorWithNorms[cluster2.dataPointIds.get(j)];
+ distance =
+ Math.min(
+ distance,
+ distanceMeasure.distance(
+ vectorWithNorm1, vectorWithNorm2));
+ }
+ }
+ break;
+ case LINKAGE_WARD:
+ int vecSize = vectorWithNorms[0].vector.size();
+ DenseVector mean1 = Vectors.dense(new double[vecSize]);
+ DenseVector mean2 = Vectors.dense(new double[vecSize]);
+
+ for (int i = 0; i < size1; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster1.dataPointIds.get(i)].vector, mean1);
+ }
+ for (int i = 0; i < size2; i++) {
+ BLAS.axpy(1.0, vectorWithNorms[cluster2.dataPointIds.get(i)].vector, mean2);
+ }
+
+ DenseVector meanMerged = mean1.clone();
+ BLAS.axpy(1.0, mean2, meanMerged);
+ BLAS.scal(1.0 / size1, mean1);
+ BLAS.scal(1.0 / size2, mean2);
+ BLAS.scal(1.0 / (size1 + size2), meanMerged);
+ double essInc =
+ size1 * BLAS.dot(mean1, mean1)
+ + size2 * BLAS.dot(mean2, mean2)
+ - (size1 + size2) * BLAS.dot(meanMerged, meanMerged);
+
+ distance = Math.sqrt(2 * essInc);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported " + LINKAGE + " type: " + linkage + ".");
+ }
+ return distance;
+ }
+
+ /** A cluster with cluster Id specified and data points that belong to this cluster. */
+ private static class Cluster {
+ private final int clusterId;
+ private final List<Integer> dataPointIds;
+
+ @SafeVarargs
+ public Cluster(
+ int clusterId, List<Integer> dataPointIds, List<Integer>... otherDataPointIds) {
Review Comment:
It might be better to write two constructors, `Cluster(int, List<Integer>)` and `Cluster(int, List<Integer>, List<Integer>)`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> K = new IntParam("k", "The max number of clusters to create.", 2);
Review Comment:
`k` has a clear meaning in Kmeans and Knn, but I'm not sure users of Agglomerative Clustering algorithm. How about changing this parameter to `numClusters` or `maxClusters`?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
Review Comment:
nit: `StreamTableEnvironment tEnv`
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
Review Comment:
Let's describe in JavaDoc the column names of the `mergeInfoTable` and their meanings.
Besides, do you think it would be better to add some method like `Table getMergeInfo()`, instead of making the merge info table the second output table? I'm not sure the current API corresponds to the syntax of Flink ML AlgoOperators. The current API also makes it difficult to connect `AgglomerativeClustering` stage with other stages in a `Pipeline`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> K = new IntParam("k", "The max number of clusters to create.", 2);
+
+ Param<Double> DISTANCE_THRESHOLD =
+ new DoubleParam(
+ "distanceThreshold",
+ "Threshold to decide whether two clusters should be merged.",
+ null);
+
+ String LINKAGE_WARD = "ward";
+ String LINKAGE_COMPLETE = "complete";
+ String LINKAGE_SINGLE = "single";
+ String LINKAGE_AVERAGE = "average";
+ /**
+ * Supported options to compute the distance between two clusters.
+ *
+ * <ul>
+ * <li>ward: difference between the sum of the variance of the two clusters and the merged
+ * one.
+ * <li>complete: the maximum distance between all observations of the two clusters.
+ * <li>single: the minimum distance between all observations of the two cluster.
+ * <li>average: the average of the distance of all observations of the two cluster.
+ * </ul>
+ */
+ Param<String> LINKAGE =
+ new StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ LINKAGE_WARD,
+ ParamValidators.inArray(
+ LINKAGE_WARD, LINKAGE_COMPLETE, LINKAGE_AVERAGE, LINKAGE_SINGLE));
+
+ Param<Boolean> COMPUTE_FULL_TREE =
+ new BooleanParam(
+ "computeFullTree",
+ "Whether computes the full tree after convergence.",
+ false,
Review Comment:
The default behavior here is different from that in sklearn. Do you think we should also support the `auto` option here?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer k,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.k = k;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (k != null && activeClusters.size() > k) || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
+ (k != null && activeClusters.size() > k)
+ || (distanceThreshold != null && distanceThreshold > minDistance);
+ }
+ }
+
+ private double computeDistanceBetweenClusters(Cluster cluster1, Cluster cluster2) {
Review Comment:
How about adding a cache here for the distance between any two data points? We can avoid re-computing the distance if there is a cache hit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r958223767
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClustering.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.common.distance.DistanceMeasure;
+import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.VectorWithNorm;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ * observation starts in its own cluster and the clusters are merged together one by one. Users can
+ * choose different strategies to merge two clusters by setting {@link
+ * AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting {@link
+ * AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Hierarchical_clustering.
+ */
+public class AgglomerativeClustering
+ implements AlgoOperator<AgglomerativeClustering>,
+ AgglomerativeClusteringParams<AgglomerativeClustering> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public AgglomerativeClustering() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer k = getK();
+ Double distanceThreshold = getDistanceThreshold();
+ Preconditions.checkArgument(
+ (k == null && distanceThreshold != null)
+ || (k != null && distanceThreshold == null),
+ "One of param k and distanceThreshold should be null.");
+
+ if (getLinkage().equals(LINKAGE_WARD)) {
+ String distanceMeasure = getDistanceMeasure();
+ Preconditions.checkArgument(
+ distanceMeasure.equals(EuclideanDistanceMeasure.NAME),
+ distanceMeasure
+ + " was provided as distance measure while linkage was ward. Ward only works with euclidean.");
+ }
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> dataStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(), Types.INT),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag =
+ new OutputTag<Tuple4<Integer, Integer, Double, Integer>>("MERGE_INFO") {};
+
+ SingleOutputStreamOperator<Row> output =
+ dataStream.transform(
+ "doLocalAgglomerativeClustering",
+ outputTypeInfo,
+ new LocalAgglomerativeClusteringOperator(
+ getFeaturesCol(),
+ getLinkage(),
+ getDistanceMeasure(),
+ getK(),
+ getDistanceThreshold(),
+ getComputeFullTree(),
+ mergeInfoOutputTag));
+ output.getTransformation().setParallelism(1);
+
+ Table outputTable = tEnv.fromDataStream(output);
+
+ DataStream<Tuple4<Integer, Integer, Double, Integer>> mergeInfo =
+ output.getSideOutput(mergeInfoOutputTag);
+ Table mergeInfoTable =
+ tEnv.fromDataStream(mergeInfo)
+ .as("clusterId1", "clusterId2", "distance", "sizeOfMergedCluster");
+
+ return new Table[] {outputTable, mergeInfoTable};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static AgglomerativeClustering load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class LocalAgglomerativeClusteringOperator extends AbstractStreamOperator<Row>
+ implements OneInputStreamOperator<Row, Row>, BoundedOneInput {
+ private final String featuresCol;
+ private final String linkage;
+ private final DistanceMeasure distanceMeasure;
+ private final Integer k;
+ private final Double distanceThreshold;
+ private final boolean computeFullTree;
+ private final OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag;
+
+ /** Cache for the input data. */
+ private List<Row> inputList;
+ /** State for the input data. */
+ private ListState<Row> inputListState;
+ /** Cluster id of each data point in inputList. */
+ private int[] clusterIds;
+ /** Precomputes the norm of each vector for performance. */
+ private VectorWithNorm[] vectorWithNorms;
+ /** Next cluster Id to be assigned. */
+ private int nextClusterId = 0;
+
+ public LocalAgglomerativeClusteringOperator(
+ String featuresCol,
+ String linkage,
+ String distanceMeasureName,
+ Integer k,
+ Double distanceThreshold,
+ boolean computeFullTree,
+ OutputTag<Tuple4<Integer, Integer, Double, Integer>> mergeInfoOutputTag) {
+ this.featuresCol = featuresCol;
+ this.linkage = linkage;
+ this.k = k;
+ this.distanceThreshold = distanceThreshold;
+ this.computeFullTree = computeFullTree;
+ this.mergeInfoOutputTag = mergeInfoOutputTag;
+
+ distanceMeasure = DistanceMeasure.getInstance(distanceMeasureName);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputListState =
+ context.getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("inputListState", Row.class));
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> input) throws Exception {
+ inputListState.add(input.getValue());
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void endInput() throws Exception {
+ inputList = IteratorUtils.toList(inputListState.get().iterator());
+ int numDataPoints = inputList.size();
+
+ // Assigns initial cluster Ids.
+ clusterIds = new int[numDataPoints];
+ for (int i = 0; i < numDataPoints; i++) {
+ clusterIds[i] = getNextClusterId();
+ }
+
+ List<Cluster> activeClusters = new ArrayList<>();
+ for (int i = 0; i < numDataPoints; i++) {
+ List<Integer> dataPointIds = new ArrayList<>();
+ dataPointIds.add(i);
+ activeClusters.add(new Cluster(i, dataPointIds));
+ }
+
+ // Precomputes vector norms for faster computation.
+ vectorWithNorms = new VectorWithNorm[inputList.size()];
+ for (int i = 0; i < numDataPoints; i++) {
+ vectorWithNorms[i] =
+ new VectorWithNorm((Vector) inputList.get(i).getField(featuresCol));
+ }
+
+ // Clustering process.
+ doClustering(activeClusters);
+
+ // Remaps the cluster Ids and output results.
+ HashMap<Integer, Integer> remappedClusterIds = new HashMap<>();
+ int cnt = 0;
+ for (int i = 0; i < clusterIds.length; i++) {
+ int clusterId = clusterIds[i];
+ if (remappedClusterIds.containsKey(clusterId)) {
+ clusterIds[i] = remappedClusterIds.get(clusterId);
+ } else {
+ clusterIds[i] = cnt;
+ remappedClusterIds.put(clusterId, cnt++);
+ }
+ }
+
+ for (int i = 0; i < numDataPoints; i++) {
+ output.collect(
+ new StreamRecord<>(Row.join(inputList.get(i), Row.of(clusterIds[i]))));
+ }
+ }
+
+ private int getNextClusterId() {
+ return nextClusterId++;
+ }
+
+ private void doClustering(List<Cluster> activeClusters) {
+ int clusterOffset1 = -1, clusterOffset2 = -1;
+ boolean clusteringRunning =
+ (k != null && activeClusters.size() > k) || (distanceThreshold != null);
+
+ while (clusteringRunning || (computeFullTree && activeClusters.size() > 1)) {
+ // Computes the distance between two clusters.
+ double minDistance = Double.MAX_VALUE;
+ for (int i = 0; i < activeClusters.size(); i++) {
+ for (int j = i + 1; j < activeClusters.size(); j++) {
+ double distance =
+ computeDistanceBetweenClusters(
+ activeClusters.get(i), activeClusters.get(j));
+ if (distance < minDistance) {
+ minDistance = distance;
+ clusterOffset1 = i;
+ clusterOffset2 = j;
+ }
+ }
+ }
+
+ // Outputs the merge info.
+ Cluster cluster1 = activeClusters.get(clusterOffset1);
+ Cluster cluster2 = activeClusters.get(clusterOffset2);
+ int clusterId1 = cluster1.clusterId;
+ int clusterId2 = cluster2.clusterId;
+ output.collect(
+ mergeInfoOutputTag,
+ new StreamRecord<>(
+ Tuple4.of(
+ Math.min(clusterId1, clusterId2),
+ Math.max(clusterId1, clusterId2),
+ minDistance,
+ cluster1.dataPointIds.size()
+ + cluster2.dataPointIds.size())));
+
+ // Merges these two clusters.
+ Cluster mergedCluster =
+ new Cluster(
+ getNextClusterId(), cluster1.dataPointIds, cluster2.dataPointIds);
+ activeClusters.set(clusterOffset1, mergedCluster);
+ activeClusters.remove(clusterOffset2);
+
+ // Updates cluster Ids for each data point if clustering is still running.
+ if (clusteringRunning) {
+ int mergedClusterId = mergedCluster.clusterId;
+ for (int dataPointId : mergedCluster.dataPointIds) {
+ clusterIds[dataPointId] = mergedClusterId;
+ }
+ }
+
+ clusteringRunning =
+ (k != null && activeClusters.size() > k)
+ || (distanceThreshold != null && distanceThreshold > minDistance);
+ }
+ }
+
+ private double computeDistanceBetweenClusters(Cluster cluster1, Cluster cluster2) {
Review Comment:
The memory overhead for caching the distance between two data points is O(n^2), where n is the number of data points.
Suppose we store the distance in a primitive double array, then the max number of data points we can deal with is Math.sqrt(Int.MAX), which is less than 100000.
A possible optimization would be cache the distance between each two clusters at the later iterations, i.e., when the number of active clusters is smaller. But we can do this in a later PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r961545453
##########
flink-ml-python/pyflink/ml/lib/clustering/agglomerativeclustering.py:
##########
@@ -0,0 +1,135 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+
+from pyflink.ml.core.param import Param, StringParam, IntParam, FloatParam, \
+ BooleanParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.clustering.common import JavaClusteringAlgoOperator
+from pyflink.ml.lib.param import HasDistanceMeasure, HasFeaturesCol, HasPredictionCol
+
+
+class _AgglomerativeClusteringParams(
+ JavaWithParams,
+ HasDistanceMeasure,
+ HasFeaturesCol,
+ HasPredictionCol
+):
+ """
+ Params for :class:`AgglomerativeClustering`.
+ """
+ NUM_CLUSTERS: Param[int] = IntParam("num_clusters",
+ "The max number of clusters to create.",
+ 2)
+
+ DISTANCE_THRESHOLD: Param[float] = \
+ FloatParam("distance_threshold",
+ "Threshold to decide whether two clusters should be merged.",
+ None)
+
+ """
+ Supported options to compute the distance between two clusters.
+ <ul>
+ <li>ward: difference between the sum of the variance of the two clusters and the merged one.
+ <li>complete: the maximum distance between all observations of the two clusters.
+ <li>single: the minimum distance between all observations of the two cluster.
+ <li>average: the average of the distance of all observations of the two cluster.
+ </ul>
+ """
+ LINKAGE: Param[str] = StringParam(
+ "linkage",
+ "Criterion for computing distance between two clusters.",
+ "ward",
+ ParamValidators.in_array(
+ ["ward", "complete", "single", "average"]))
+
+ COMPUTE_FULL_TREE: Param[bool] = BooleanParam(
+ "compute_full_tree",
+ "Whether computes the full tree after convergence.",
+ False,
+ ParamValidators.not_null())
+
+ def __init__(self, java_params):
+ super(_AgglomerativeClusteringParams, self).__init__(java_params)
+
+ def set_num_clusters(self, value: int):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.NUM_CLUSTERS, value))
+
+ def get_num_clusters(self) -> int:
+ return self.get(self.NUM_CLUSTERS)
+
+ def set_distance_threshold(self, value: float):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.DISTANCE_THRESHOLD, value))
+
+ def get_distance_threshold(self) -> float:
+ return self.get(self.DISTANCE_THRESHOLD)
+
+ def set_linkage(self, value: str):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.LINKAGE, value))
+
+ def get_linkage(self) -> str:
+ return self.get(self.LINKAGE)
+
+ def set_compute_full_tree(self, value: bool):
+ return typing.cast(_AgglomerativeClusteringParams, self.set(self.COMPUTE_FULL_TREE, value))
+
+ def get_compute_full_tree(self) -> bool:
+ return self.get(self.COMPUTE_FULL_TREE)
+
+ @property
+ def num_clusters(self):
+ return self.get_num_clusters()
+
+ @property
+ def distance_threshold(self):
+ return self.get_distance_threshold()
+
+ @property
+ def linkage(self):
+ return self.get_linkage()
+
+ @property
+ def compute_full_tree(self):
+ return self.get_compute_full_tree()
+
+
+class AgglomerativeClustering(JavaClusteringAlgoOperator, _AgglomerativeClusteringParams):
+ """
+ An AlgoOperator that performs a hierarchical clustering using a bottom up approach. Each
+ observation starts in its own cluster and the clusters are merged together one by one.
+ Users can choose different strategies to merge two clusters by setting
+ {@link AgglomerativeClusteringParams#LINKAGE} and different distance measure by setting
+ {@link AgglomerativeClusteringParams#DISTANCE_MEASURE}.
+
+ <p>The output contains two tables. The first one assigns one cluster Id for each data point.
+ The second one contains the information of merging two clusters at each step. The data format
+ of the merging information is (clusterId1, clusterId2, distance, sizeOfMergedCluster).
Review Comment:
Thanks for the comment. An example script to visualize the merge info is added [1]. It is also mentioned in the user doc [2].
[1] flink-ml-dist/src/main/flink-ml-bin/bin/agglomerativeclustering-visualize.py
[2] docs/content/docs/operators/clustering/agglomerativeclustering.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r958225992
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> K = new IntParam("k", "The max number of clusters to create.", 2);
Review Comment:
Thanks for the comments. I agree that `numClusters` is more explainable and have renamed the param.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #148: [FLINK-28906] Add AlgoOperator for AgglomerativeClustering
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on code in PR #148:
URL: https://github.com/apache/flink-ml/pull/148#discussion_r958225992
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/agglomerativeclustering/AgglomerativeClusteringParams.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.clustering.agglomerativeclustering;
+
+import org.apache.flink.ml.common.param.HasDistanceMeasure;
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+
+/**
+ * Params of {@link AgglomerativeClustering}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface AgglomerativeClusteringParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Integer> K = new IntParam("k", "The max number of clusters to create.", 2);
Review Comment:
`k` here is exactly the same with that in Kmeans. I didn't extract it as shared param because `k` is used in many different cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org