You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "A-Forward-Scholar (via GitHub)" <gi...@apache.org> on 2023/08/31 10:01:16 UTC
[GitHub] [flink-ml] A-Forward-Scholar opened a new pull request, #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
A-Forward-Scholar opened a new pull request, #253:
URL: https://github.com/apache/flink-ml/pull/253
<!--
*Thank you very much for contributing to Apache Flink ML - we are happy that you want to help us improve Flink ML. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to one [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] Title of the pull request`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request adds the implementation of Naive Bayes algorithm.)*
## Brief change log
*(for example:)*
- *Adds Transformer and Estimator implementation of Naive Bayes in Java and Python*
- *Adds examples and documentations of Naive Bayes*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1318011604
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForest}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T> extends IsolationForestModelParams<T> {
+ Param<Integer> MAX_SAMPLES =
+ new IntParam(
+ "maxSamples",
+ "The number of samplesData to train and its max value is preferably 256.",
+ 256);
+
+ Param<Double> MAX_FEATURES =
+ new DoubleParam(
+ "maxFeatures",
+ "The number of features used to train each tree and it is treated as a fraction in the range (0, 1.0].",
+ 1.0);
+
+ default int getNumTrees() {
Review Comment:
Yes, teacher, I have deleted him in second pr which will overwrite the first 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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314378789
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
Review Comment:
Only used in calculating center0 and center1, I will rethink it.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1318012250
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/ITree.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.Random;
+
+/** Construct isolation tree. */
+public class ITree implements Serializable {
+ public final int attributeIndex;
+ public final double splitAttributeValue;
+ public ITree leftTree;
+ public ITree rightTree;
+ public int currentHeight;
+ public int leafNodesNum;
+
+ public ITree(int attributeIndex, double splitAttributeValue) {
+ this.attributeIndex = attributeIndex;
+ this.splitAttributeValue = splitAttributeValue;
+ this.leftTree = null;
+ this.rightTree = null;
+ this.currentHeight = 0;
+ this.leafNodesNum = 1;
+ }
+
+ public static ITree generateIsolationTree(
+ DenseVector[] samplesData,
+ int currentHeight,
+ int limitHeight,
+ Random randomState,
+ int[] featureIndices) {
+ int n = samplesData.length;
+ ITree isolationTree;
+ if (samplesData.length == 0) {
+ return null;
+ } else if (samplesData.length == 1 || currentHeight >= limitHeight) {
+ isolationTree = new ITree(0, samplesData[0].get(0));
+ isolationTree.currentHeight = currentHeight;
+ isolationTree.leafNodesNum = samplesData.length;
+ return isolationTree;
+ }
+ boolean flag = true;
+ for (int i = 1; i < n; i++) {
+ if (!samplesData[i].equals(samplesData[i - 1])) {
+ flag = false;
+ break;
+ }
+ }
+ if (flag) {
+ isolationTree = new ITree(0, samplesData[0].get(0));
+ isolationTree.currentHeight = currentHeight;
+ isolationTree.leafNodesNum = samplesData.length;
+ return isolationTree;
+ }
+
+ Tuple2<Integer, Double> tuple2 =
+ getRandomFeatureToSplit(samplesData, randomState, featureIndices);
+ int attributeIndex = tuple2.f0;
+ double splitAttributeValue = tuple2.f1;
+
+ int leftNodesNum = 0;
+ int rightNodesNum = 0;
+ for (DenseVector datum : samplesData) {
+ if (datum.get(attributeIndex) < splitAttributeValue) {
+ leftNodesNum++;
+ } else {
+ rightNodesNum++;
+ }
+ }
+
+ DenseVector[] leftSamples = new DenseVector[leftNodesNum];
+ DenseVector[] rightSamples = new DenseVector[rightNodesNum];
+ int l = 0, r = 0;
+ for (DenseVector samplesDatum : samplesData) {
+ if (samplesDatum.get(attributeIndex) < splitAttributeValue) {
+ leftSamples[l++] = samplesDatum;
+ } else {
+ rightSamples[r++] = samplesDatum;
+ }
+ }
+
+ ITree root = new ITree(attributeIndex, splitAttributeValue);
+ root.currentHeight = currentHeight;
+ root.leafNodesNum = samplesData.length;
+ root.leftTree =
+ generateIsolationTree(
+ leftSamples, currentHeight + 1, limitHeight, randomState, featureIndices);
+ root.rightTree =
+ generateIsolationTree(
+ rightSamples, currentHeight + 1, limitHeight, randomState, featureIndices);
+
+ return root;
+ }
+
+ private static Tuple2<Integer, Double> getRandomFeatureToSplit(
+ DenseVector[] samplesData, Random randomState, int[] featureIndices) {
+ int attributeIndex = featureIndices[randomState.nextInt(featureIndices.length)];
+
+ double maxValue = samplesData[0].get(attributeIndex);
+ double minValue = samplesData[0].get(attributeIndex);
+ for (int i = 1; i < samplesData.length; i++) {
+ minValue = Math.min(minValue, samplesData[i].get(attributeIndex));
+ maxValue = Math.max(maxValue, samplesData[i].get(attributeIndex));
+ }
+ double splitAttributeValue = (maxValue - minValue) * randomState.nextDouble() + minValue;
+
+ return Tuple2.of(attributeIndex, splitAttributeValue);
+ }
+
+ public static double calculatePathLength(DenseVector sampleData, ITree isolationTree)
+ throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = isolationTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex) == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ assert tmpITree != null;
Review Comment:
Ok, I will delete it.
--
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 #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1313857237
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T>, HasWindows<T> {
+ Param<Integer> TREES_NUMBER =
+ new IntParam("treesNumber", "The max number of trees to create.", 2);
+
+ Param<Integer> ITERS =
Review Comment:
Would it be simpler to re-use HasMaxIter?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
Review Comment:
Can we mark these variables `final` if possible?
`final` variables are immutable and more readable in general.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestModelData.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.ml.anomalydetection.isolationforest.IsolationForest.IForest;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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 java.io.EOFException;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Model data of {@link IsolationForestModel}.
+ *
+ * <p>This class also provides methods to convert model data from Table to Datastream, and classes
+ * to save/load model data.
+ */
+public class IsolationForestModelData {
+
+ public IForest iForest;
Review Comment:
Can we move the definition of the data structure which represents the model data from `IsolationForest.java` to `IsolationForestModelData.java`? This can make the code a bit more readable by separating data structure from the training/processing logic.
This can also make the code more consistent with the existing code style.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T>, HasWindows<T> {
+ Param<Integer> TREES_NUMBER =
+ new IntParam("treesNumber", "The max number of trees to create.", 2);
Review Comment:
It seems that `IForest#iTreeList` would contain exactly the given number of trees. Would it be more accurate to specify "The number of trees" in the doc?
If yes, it might be better to name this parameter `numTrees` for consistency with the `numTrees` parameter used by `org.apache.spark.mllib.tree.RandomForest` in Spark ML.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
+ public double splitAttributeValue;
+ public ITree leftTree, rightTree;
Review Comment:
Can we declare one variable at a line for consistency with the existing code style?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
Review Comment:
Would it be better to let `IsolationForestParams` extend `IsolationForestModelParams` for consistency with e.g. how `LogisticRegressionParams` extends `LogisticRegressionModelParams`?
This change can make it explicit that the estimator parameters covers all the corresponding model's parameter.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestModel.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.anomalydetection.isolationforest.IsolationForest.ITree;
+import org.apache.flink.ml.api.Model;
+import org.apache.flink.ml.common.broadcast.BroadcastUtils;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+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.ml.util.RowUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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.Preconditions;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A Model which detection anomaly data using the model data computed by {@link IsolationForest}.
+ */
+public class IsolationForestModel
+ implements Model<IsolationForestModel>, IsolationForestParams<IsolationForestModel> {
Review Comment:
Would it be better to replace `IsolationForestParams` with `IsolationForestModelParams` in this line?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
Review Comment:
Can you update PR so that it does not contain Chinese?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
+ public double splitAttributeValue;
+ public ITree leftTree, rightTree;
+ public int currentHeight;
+ public int leafNodesNum;
+
+ public ITree() {}
+
+ public ITree(int attributeIndex, double splitAttributeValue) {
+ this.attributeIndex = attributeIndex;
+ this.splitAttributeValue = splitAttributeValue;
+ this.leftTree = null;
+ this.rightTree = null;
+ this.currentHeight = 0;
+ this.leafNodesNum = 1;
+ }
+
+ public static ITree createITree(
Review Comment:
If `ITree` is used to describe model data, it might be better to move the logic of computing/processing `ITree` out of this class.
Ideally, the description of the model data should be put in the `IsolationForestModelData.java` and the training logic should be put in the corresponding estimator class.
Same for `IForest`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
Review Comment:
Given that `IForest` is used as the model data, it seems a bit weird to have `iters` in the model data.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314378326
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
Review Comment:
The consideration in this aspect is that the follow-up needs to rely on IForest to calculate the distance from center0 and center1, so as to judge which point is closer to this point. IForest and the maximum number of averages retained by ITreeList need to be used.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314372168
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
Review Comment:
ok, I will fix it.
--
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] zhaozijun109 commented on pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on PR #253:
URL: https://github.com/apache/flink-ml/pull/253#issuecomment-1707585256
@lindong28 Please review again teacher，thank you.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1318013033
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/ITree.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.Random;
+
+/** Construct isolation tree. */
+public class ITree implements Serializable {
+ public final int attributeIndex;
+ public final double splitAttributeValue;
+ public ITree leftTree;
+ public ITree rightTree;
+ public int currentHeight;
Review Comment:
Sounds good for me, I will make it final.
--
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] zhaozijun109 commented on pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on PR #253:
URL: https://github.com/apache/flink-ml/pull/253#issuecomment-1709386231
> Thanks for your contribution and the PR update! I left some comments below.
>
> Sorry, I might not have time to actively review this PR later. I will ask if some of my colleagues can help review this PR.
>
> The PR currently fails compilation due to the following error in the log [1]. I don't think it is related to either pyflink or beam. But I probably won't be able to dig into this anytime soon.
>
> ```
> Error: Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.0:compile (default-compile) on project flink-ml-lib-1.17: Compilation failure: Compilation failure:
> Error: /home/runner/work/flink-ml/flink-ml/flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:[302,21] cannot infer type arguments for org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache<>
> Error: reason: cannot infer type-variable(s) T
> Error: (actual and formal argument lists differ in length)
> ```
>
> [1] https://github.com/apache/flink-ml/actions/runs/6092176700/job/16543229469?pr=253
This is due to this PR (https://github.com/apache/flink-ml/commit/ebdf362664283bacad476126ee18c4059593950a) induced，I will fix the problem.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314374284
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T>, HasWindows<T> {
+ Param<Integer> TREES_NUMBER =
+ new IntParam("treesNumber", "The max number of trees to create.", 2);
Review Comment:
OK, I will rename the parameter.
--
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] zhaozijun109 commented on pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on PR #253:
URL: https://github.com/apache/flink-ml/pull/253#issuecomment-1709379802
> Thanks for your contribution and the PR update! I left some comments below.
>
> Sorry, I might not have time to actively review this PR later. I will ask if some of my colleagues can help review this PR.
>
> The PR currently fails compilation due to the following error in the log [1]. I don't think it is related to either pyflink or beam. But I probably won't be able to dig into this anytime soon.
>
> ```
> Error: Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.0:compile (default-compile) on project flink-ml-lib-1.17: Compilation failure: Compilation failure:
> Error: /home/runner/work/flink-ml/flink-ml/flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:[302,21] cannot infer type arguments for org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache<>
> Error: reason: cannot infer type-variable(s) T
> Error: (actual and formal argument lists differ in length)
> ```
>
> [1] https://github.com/apache/flink-ml/actions/runs/6092176700/job/16543229469?pr=253
Teacher, this problem seems to be ListStateWithCache class. I am not sure if anyone has submitted the relevant PR. I will rebuild the branch based on the latest master, test it and resubmit it.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314376830
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestModelData.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.ml.anomalydetection.isolationforest.IsolationForest.IForest;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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 java.io.EOFException;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Model data of {@link IsolationForestModel}.
+ *
+ * <p>This class also provides methods to convert model data from Table to Datastream, and classes
+ * to save/load model data.
+ */
+public class IsolationForestModelData {
+
+ public IForest iForest;
Review Comment:
Thanks for your suggestion, I will modify it.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314373790
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
+ extends HasDistanceMeasure<T>, HasFeaturesCol<T>, HasPredictionCol<T>, HasWindows<T> {
+ Param<Integer> TREES_NUMBER =
+ new IntParam("treesNumber", "The max number of trees to create.", 2);
+
+ Param<Integer> ITERS =
Review Comment:
Maybe, I will try. I haven't seen HasMaxIter before.
--
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] A-Forward-Scholar commented on pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "A-Forward-Scholar (via GitHub)" <gi...@apache.org>.
A-Forward-Scholar commented on PR #253:
URL: https://github.com/apache/flink-ml/pull/253#issuecomment-1702040771
Hello teachers, this problem seems that the beam package in pyflink has changed. How should I solve this problem?
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314375629
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestModel.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.anomalydetection.isolationforest.IsolationForest.ITree;
+import org.apache.flink.ml.api.Model;
+import org.apache.flink.ml.common.broadcast.BroadcastUtils;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+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.ml.util.RowUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+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.Preconditions;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A Model which detection anomaly data using the model data computed by {@link IsolationForest}.
+ */
+public class IsolationForestModel
+ implements Model<IsolationForestModel>, IsolationForestParams<IsolationForestModel> {
Review Comment:
Same as above.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314375469
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+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.common.param.HasWindows;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForestModel}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T>
Review Comment:
Yes, because I want to continue to try an online version later, I am not sure whether this will be used, but I will remove it first, teacher.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314379507
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
+ public double splitAttributeValue;
+ public ITree leftTree, rightTree;
+ public int currentHeight;
+ public int leafNodesNum;
+
+ public ITree() {}
+
+ public ITree(int attributeIndex, double splitAttributeValue) {
+ this.attributeIndex = attributeIndex;
+ this.splitAttributeValue = splitAttributeValue;
+ this.leftTree = null;
+ this.rightTree = null;
+ this.currentHeight = 0;
+ this.leafNodesNum = 1;
+ }
+
+ public static ITree createITree(
Review Comment:
OK, thanks for your advice and review teacher, I will fix them one by one.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314376355
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
+ public double splitAttributeValue;
+ public ITree leftTree, rightTree;
Review Comment:
Ok.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1314376152
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForest.java:
##########
@@ -0,0 +1,611 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.iteration.DataStreamList;
+import org.apache.flink.iteration.IterationBody;
+import org.apache.flink.iteration.IterationBodyResult;
+import org.apache.flink.iteration.IterationConfig;
+import org.apache.flink.iteration.IterationListener;
+import org.apache.flink.iteration.Iterations;
+import org.apache.flink.iteration.ReplayableDataStreamList;
+import org.apache.flink.iteration.datacache.nonkeyed.ListStateWithCache;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.iteration.ForwardInputsOfLastRound;
+import org.apache.flink.ml.common.iteration.TerminateOnMaxIter;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.TwoInputStreamOperator;
+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.Collector;
+import org.apache.flink.util.OutputTag;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/**
+ * An Estimator which implements the Isolation Forest algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/Isolation_forest.
+ */
+public class IsolationForest
+ implements Estimator<IsolationForest, IsolationForestModel>,
+ IsolationForestParams<IsolationForest> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public IsolationForest() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public IsolationForestModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Integer treesNumber = getTreesNumber();
+ Integer iters = getIters();
+ Preconditions.checkArgument(
+ treesNumber != null || treesNumber > 0, "Param treesNumber is illegal.");
+ Preconditions.checkArgument(iters != null || iters > 0, "Param iters is illegal.");
+ IForest iForest = new IForest(treesNumber, iters);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+
+ DataStream<DenseVector[]> points =
+ tEnv.toDataStream(inputs[0]).map(new FormatDataMapFunction(getFeaturesCol()));
+
+ DataStream<IForest> initModelData =
+ selectRandomSample1(points).map(new InitModelData(iForest)).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ Iterations.iterateBoundedStreamsUntilTermination(
+ DataStreamList.of(initModelData),
+ ReplayableDataStreamList.notReplay(points),
+ IterationConfig.newBuilder()
+ .setOperatorLifeCycle(
+ IterationConfig.OperatorLifeCycle.ALL_ROUND)
+ .build(),
+ new IsolationForestIterationBody(iters))
+ .get(0);
+
+ Table finalModelDataTable = tEnv.fromDataStream(finalModelData);
+ IsolationForestModel model = new IsolationForestModel().setModelData(finalModelDataTable);
+ ParamUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static IsolationForestModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ private static class FormatDataMapFunction implements MapFunction<Row, DenseVector[]> {
+ private final String featuresCol;
+ private List<DenseVector> list;
+
+ public FormatDataMapFunction(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public DenseVector[] map(Row row) throws Exception {
+ list = new ArrayList<>(256);
+ DenseVector denseVector = ((Vector) row.getField(featuresCol)).toDense();
+ list.add(denseVector);
+ return list.toArray(new DenseVector[0]);
+ }
+ }
+
+ private static DataStream<DenseVector[]> selectRandomSample1(
+ DataStream<DenseVector[]> samplesData) {
+ DataStream<DenseVector[]> resultStream =
+ DataStreamUtils.mapPartition(
+ DataStreamUtils.sample(samplesData, 256, System.currentTimeMillis()),
+ (MapPartitionFunction<DenseVector[], DenseVector[]>)
+ (iterable, collector) -> {
+ Iterator<DenseVector[]> samplesDataIterator =
+ iterable.iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ list.addAll(Arrays.asList(samplesDataIterator.next()));
+ }
+ collector.collect(list.toArray(new DenseVector[0]));
+ },
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE));
+ resultStream.getTransformation().setParallelism(1);
+ return resultStream;
+ }
+
+ private static class InitModelData implements MapFunction<DenseVector[], IForest> {
+ private final IForest iForest;
+
+ private InitModelData(IForest iForest) {
+ this.iForest = iForest;
+ }
+
+ @Override
+ public IForest map(DenseVector[] denseVectors) throws Exception {
+ iForest.createIForest(denseVectors);
+ DenseVector scores = iForest.calculateAnomalyScore(denseVectors);
+ iForest.classifyByCluster(scores);
+ return iForest;
+ }
+ }
+
+ private static class IsolationForestIterationBody implements IterationBody {
+ private final Integer iters;
+
+ public IsolationForestIterationBody(Integer iters) {
+ this.iters = iters;
+ }
+
+ @Override
+ public IterationBodyResult process(
+ DataStreamList variableStreams, DataStreamList dataStreams) {
+ DataStream<IForest> centersData = variableStreams.get(0);
+ DataStream<DenseVector[]> samplesData = dataStreams.get(0);
+ final OutputTag<IForest> modelDataOutputTag =
+ new OutputTag<IForest>("IsolationForest") {};
+
+ SingleOutputStreamOperator terminationCriteria =
+ centersData.flatMap(new TerminateOnMaxIter(iters));
+
+ DataStream<IForest> centers =
+ samplesData
+ .connect(centersData.broadcast())
+ .transform(
+ "CentersUpdateAccumulator",
+ TypeInformation.of(IForest.class),
+ new CentersUpdateAccumulator(modelDataOutputTag));
+
+ DataStream<IsolationForestModelData> newModelData =
+ centers.countWindowAll(centers.getParallelism())
+ .reduce(
+ new ReduceFunction<IForest>() {
+ @Override
+ public IForest reduce(IForest iForest1, IForest iForest2)
+ throws Exception {
+ if (iForest2.center0 == null
+ || iForest2.center1 == null) {
+ return iForest1;
+ }
+ return iForest2;
+ }
+ })
+ .flatMap(
+ new FlatMapFunction<IForest, IsolationForestModelData>() {
+ @Override
+ public void flatMap(
+ IForest iForest,
+ Collector<IsolationForestModelData> collector)
+ throws Exception {
+ if (iForest.center0 != null
+ && iForest.center1 != null) {
+ collector.collect(
+ new IsolationForestModelData(iForest));
+ }
+ }
+ });
+
+ DataStream<IForest> newCenters = newModelData.map(x -> x.iForest).setParallelism(1);
+
+ DataStream<IsolationForestModelData> finalModelData =
+ newModelData.flatMap(new ForwardInputsOfLastRound<>());
+
+ return new IterationBodyResult(
+ DataStreamList.of(newCenters),
+ DataStreamList.of(finalModelData),
+ terminationCriteria);
+ }
+ }
+
+ private static class CentersUpdateAccumulator extends AbstractStreamOperator<IForest>
+ implements TwoInputStreamOperator<DenseVector[], IForest, IForest>,
+ IterationListener<IForest> {
+ private final OutputTag<IForest> modelDataOutputTag;
+
+ private ListStateWithCache<DenseVector[]> samplesData;
+
+ private ListState<IForest> samplesDataCenter;
+
+ private ListStateWithCache<DenseVector[]> samplesDataScores;
+
+ public CentersUpdateAccumulator(OutputTag<IForest> modelDataOutputTag) {
+ this.modelDataOutputTag = modelDataOutputTag;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+
+ samplesDataCenter =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<IForest>(
+ "centers", TypeInformation.of(IForest.class)));
+
+ samplesData =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+
+ samplesDataScores =
+ new ListStateWithCache<>(
+ Types.OBJECT_ARRAY(DenseVectorTypeInfo.INSTANCE)
+ .createSerializer(getExecutionConfig()),
+ getContainingTask(),
+ getRuntimeContext(),
+ context,
+ config.getOperatorID());
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ samplesData.snapshotState(context);
+ }
+
+ @Override
+ public void processElement1(StreamRecord<DenseVector[]> streamRecord) throws Exception {
+ samplesData.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processElement2(StreamRecord<IForest> streamRecord) throws Exception {
+ Preconditions.checkState(!samplesDataCenter.get().iterator().hasNext());
+ samplesDataCenter.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void onEpochWatermarkIncremented(
+ int epochWatermark, Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ Iterator<DenseVector[]> samplesDataIterator = samplesData.get().iterator();
+ List<DenseVector> list = new ArrayList<>();
+ while (samplesDataIterator.hasNext()) {
+ DenseVector[] sampleData = samplesDataIterator.next();
+ list.add(centers.calculateAnomalyScore(sampleData));
+ }
+ DenseVector[] scores = list.toArray(new DenseVector[0]);
+ samplesDataScores.add(scores);
+
+ collector.collect(samplesDataCenter.get().iterator().next());
+ samplesDataCenter.clear();
+ }
+
+ @Override
+ public void onIterationTerminated(Context context, Collector<IForest> collector)
+ throws Exception {
+ IForest centers =
+ Objects.requireNonNull(
+ OperatorStateUtils.getUniqueElement(samplesDataCenter, "centers")
+ .orElse(null));
+ double centers0Sum1 = 0.0, centers1Sum1 = 0.0, centers0Sum2 = 0.0, centers1Sum2 = 0.0;
+ int size1 = 0, size2 = 0;
+ Iterator<DenseVector[]> samplesDataScoresIterator = samplesDataScores.get().iterator();
+ while (samplesDataScoresIterator.hasNext()) {
+ for (DenseVector denseVector : samplesDataScoresIterator.next()) {
+ DenseVector denseVector1 = centers.classifyByCluster(denseVector);
+ centers0Sum1 += denseVector1.get(0);
+ centers1Sum1 += denseVector1.get(1);
+ size1++;
+ }
+ centers0Sum2 = centers0Sum1 / size1;
+ centers1Sum2 = centers1Sum1 / size1;
+ size2++;
+ }
+
+ centers.center0 = centers0Sum2 / size2;
+ centers.center1 = centers1Sum2 / size2;
+
+ context.output(modelDataOutputTag, centers);
+
+ samplesDataCenter.clear();
+ samplesDataScores.clear();
+ samplesData.clear();
+ }
+ }
+
+ /** Construct isolation forest. */
+ public static class IForest implements Serializable {
+ public int treesNumber;
+ public int iters;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest() {}
+
+ public IForest(int treesNumber, int iters) {
+ this.iters = iters;
+ this.treesNumber = treesNumber;
+ this.iTreeList = new ArrayList<>();
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ private void createIForest(DenseVector[] samplesData) throws Exception {
+ this.subSamplesSize = Math.min(256, samplesData.length);
+
+ // 限制高度(向上取整)
+ int limitHeight = (int) Math.ceil(Math.log(subSamplesSize) / Math.log(2));
+
+ int rows = samplesData.length;
+
+ Random random = new Random(System.currentTimeMillis());
+ for (int i = 0; i < this.treesNumber; i++) {
+ DenseVector[] subSamples = new DenseVector[subSamplesSize];
+ for (int j = 0; j < subSamplesSize; j++) {
+ int r = random.nextInt(rows);
+ subSamples[j] = samplesData[r];
+ }
+ ITree iTree = ITree.createITree(subSamples, 0, limitHeight);
+ this.iTreeList.add(iTree);
+ }
+ }
+
+ private DenseVector calculateAnomalyScore(DenseVector[] samplesData) throws Exception {
+ int n = samplesData.length;
+
+ DenseVector scores = new DenseVector(n);
+ for (int i = 0; i < n; i++) {
+ double pathLengthSum = 0;
+ for (ITree iTree : iTreeList) {
+ pathLengthSum += calculatePathLength(samplesData[i], iTree);
+ }
+
+ double pathLengthAvg = pathLengthSum / iTreeList.size();
+ double cn = calculateCn(subSamplesSize);
+ double index = pathLengthAvg / cn;
+ scores.set(i, Math.pow(2, -index));
+ }
+ return scores;
+ }
+
+ private double calculatePathLength(DenseVector sampleData, ITree iTree) throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = iTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex)
+ == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ return pathLength + calculateCn(tmpITree.leafNodesNum);
+ }
+
+ private double calculateCn(double n) {
+ if (n <= 1) {
+ return 0;
+ }
+ return 2.0 * (Math.log(n - 1.0) + 0.5772156649015329) - 2.0 * (n - 1.0) / n;
+ }
+
+ private DenseVector classifyByCluster(DenseVector scores) {
+ int scoresSize = scores.size();
+ this.center0 = scores.get(0); // Cluster center of abnormal
+ this.center1 = scores.get(0); // Cluster center of normal
+
+ for (int p = 1; p < scores.size(); p++) {
+ if (scores.get(p) > center0) {
+ center0 = scores.get(p);
+ }
+
+ if (scores.get(p) < center1) {
+ center1 = scores.get(p);
+ }
+ }
+
+ int cnt0, cnt1;
+ double diff0, diff1;
+ int[] labels = new int[scoresSize];
+
+ for (int i = 0; i < iters; i++) {
+ cnt0 = 0;
+ cnt1 = 0;
+
+ for (int j = 0; j < scoresSize; j++) {
+ diff0 = Math.abs(scores.get(j) - center0);
+ diff1 = Math.abs(scores.get(j) - center1);
+
+ if (diff0 < diff1) {
+ labels[j] = 0;
+ cnt0++;
+ } else {
+ labels[j] = 1;
+ cnt1++;
+ }
+ }
+
+ diff0 = center0;
+ diff1 = center1;
+
+ center0 = 0.0;
+ center1 = 0.0;
+ for (int k = 0; k < scoresSize; k++) {
+ if (labels[k] == 0) {
+ center0 += scores.get(k);
+ } else {
+ center1 += scores.get(k);
+ }
+ }
+
+ center0 /= cnt0;
+ center1 /= cnt1;
+
+ if (center0 - diff0 <= 1e-6 && center1 - diff1 <= 1e-6) {
+ break;
+ }
+ }
+ return new DenseVector(new double[] {center0, center1});
+ }
+ }
+
+ /** Construct isolation tree. */
+ public static class ITree implements Serializable {
+ public int attributeIndex;
Review Comment:
Ok, I will recheck all parameters.
--
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 #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1317198364
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/ITree.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.Random;
+
+/** Construct isolation tree. */
+public class ITree implements Serializable {
+ public final int attributeIndex;
+ public final double splitAttributeValue;
+ public ITree leftTree;
+ public ITree rightTree;
+ public int currentHeight;
+ public int leafNodesNum;
+
+ public ITree(int attributeIndex, double splitAttributeValue) {
+ this.attributeIndex = attributeIndex;
+ this.splitAttributeValue = splitAttributeValue;
+ this.leftTree = null;
+ this.rightTree = null;
+ this.currentHeight = 0;
+ this.leafNodesNum = 1;
+ }
+
+ public static ITree generateIsolationTree(
+ DenseVector[] samplesData,
+ int currentHeight,
+ int limitHeight,
+ Random randomState,
+ int[] featureIndices) {
+ int n = samplesData.length;
+ ITree isolationTree;
+ if (samplesData.length == 0) {
+ return null;
+ } else if (samplesData.length == 1 || currentHeight >= limitHeight) {
+ isolationTree = new ITree(0, samplesData[0].get(0));
+ isolationTree.currentHeight = currentHeight;
+ isolationTree.leafNodesNum = samplesData.length;
+ return isolationTree;
+ }
+ boolean flag = true;
+ for (int i = 1; i < n; i++) {
+ if (!samplesData[i].equals(samplesData[i - 1])) {
+ flag = false;
+ break;
+ }
+ }
+ if (flag) {
+ isolationTree = new ITree(0, samplesData[0].get(0));
+ isolationTree.currentHeight = currentHeight;
+ isolationTree.leafNodesNum = samplesData.length;
+ return isolationTree;
+ }
+
+ Tuple2<Integer, Double> tuple2 =
+ getRandomFeatureToSplit(samplesData, randomState, featureIndices);
+ int attributeIndex = tuple2.f0;
+ double splitAttributeValue = tuple2.f1;
+
+ int leftNodesNum = 0;
+ int rightNodesNum = 0;
+ for (DenseVector datum : samplesData) {
+ if (datum.get(attributeIndex) < splitAttributeValue) {
+ leftNodesNum++;
+ } else {
+ rightNodesNum++;
+ }
+ }
+
+ DenseVector[] leftSamples = new DenseVector[leftNodesNum];
+ DenseVector[] rightSamples = new DenseVector[rightNodesNum];
+ int l = 0, r = 0;
+ for (DenseVector samplesDatum : samplesData) {
+ if (samplesDatum.get(attributeIndex) < splitAttributeValue) {
+ leftSamples[l++] = samplesDatum;
+ } else {
+ rightSamples[r++] = samplesDatum;
+ }
+ }
+
+ ITree root = new ITree(attributeIndex, splitAttributeValue);
+ root.currentHeight = currentHeight;
+ root.leafNodesNum = samplesData.length;
+ root.leftTree =
+ generateIsolationTree(
+ leftSamples, currentHeight + 1, limitHeight, randomState, featureIndices);
+ root.rightTree =
+ generateIsolationTree(
+ rightSamples, currentHeight + 1, limitHeight, randomState, featureIndices);
+
+ return root;
+ }
+
+ private static Tuple2<Integer, Double> getRandomFeatureToSplit(
+ DenseVector[] samplesData, Random randomState, int[] featureIndices) {
+ int attributeIndex = featureIndices[randomState.nextInt(featureIndices.length)];
+
+ double maxValue = samplesData[0].get(attributeIndex);
+ double minValue = samplesData[0].get(attributeIndex);
+ for (int i = 1; i < samplesData.length; i++) {
+ minValue = Math.min(minValue, samplesData[i].get(attributeIndex));
+ maxValue = Math.max(maxValue, samplesData[i].get(attributeIndex));
+ }
+ double splitAttributeValue = (maxValue - minValue) * randomState.nextDouble() + minValue;
+
+ return Tuple2.of(attributeIndex, splitAttributeValue);
+ }
+
+ public static double calculatePathLength(DenseVector sampleData, ITree isolationTree)
+ throws Exception {
+ double pathLength = -1;
+ ITree tmpITree = isolationTree;
+ while (tmpITree != null) {
+ pathLength += 1;
+ if (tmpITree.leftTree == null
+ || tmpITree.rightTree == null
+ || sampleData.get(tmpITree.attributeIndex) == tmpITree.splitAttributeValue) {
+ break;
+ } else if (sampleData.get(tmpITree.attributeIndex) < tmpITree.splitAttributeValue) {
+ tmpITree = tmpITree.leftTree;
+ } else {
+ tmpITree = tmpITree.rightTree;
+ }
+ }
+
+ assert tmpITree != null;
Review Comment:
In general we only use `assert` in unit tests.
Given that we will have NullPointerException in the line below if `tmpITree == null`, it seems simpler to just remove this line.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IsolationForestParams.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params of {@link IsolationForest}.
+ *
+ * @param <T> The class of this instance.
+ */
+public interface IsolationForestParams<T> extends IsolationForestModelParams<T> {
+ Param<Integer> MAX_SAMPLES =
+ new IntParam(
+ "maxSamples",
+ "The number of samplesData to train and its max value is preferably 256.",
+ 256);
+
+ Param<Double> MAX_FEATURES =
+ new DoubleParam(
+ "maxFeatures",
+ "The number of features used to train each tree and it is treated as a fraction in the range (0, 1.0].",
+ 1.0);
+
+ default int getNumTrees() {
Review Comment:
Given that this method is already defined in `IsolationForestModelParams`, should we remove it here?
Same for `setNumTrees()`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/ITree.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.Random;
+
+/** Construct isolation tree. */
+public class ITree implements Serializable {
+ public final int attributeIndex;
+ public final double splitAttributeValue;
+ public ITree leftTree;
+ public ITree rightTree;
+ public int currentHeight;
Review Comment:
Given that we only mutate `currentHeight` right after it is constructed, it should be possible and better to make it `final`.
Same for the other three non-final variables.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IForest.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+/** Construct isolation forest. */
+public class IForest implements Serializable {
+ public final int numTrees;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest(int numTrees) {
+ this.numTrees = numTrees;
+ this.iTreeList = new ArrayList<>(256);
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ public void generateIsolationForest(DenseVector[] samplesData, int[] featureIndices) {
+ int n = samplesData.length;
+ this.subSamplesSize = Math.min(256, n);
Review Comment:
nits: would it be more consistent with existing code to use `subSamplesSize` directly here?
Same for other usages of `this.*` outside constructor.
--
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] zhaozijun109 commented on a diff in pull request #253: [FLINK-33003] Support isolation forest algorithm in Flink ML
Posted by "zhaozijun109 (via GitHub)" <gi...@apache.org>.
zhaozijun109 commented on code in PR #253:
URL: https://github.com/apache/flink-ml/pull/253#discussion_r1318012024
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/anomalydetection/isolationforest/IForest.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.anomalydetection.isolationforest;
+
+import org.apache.flink.ml.linalg.DenseVector;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+/** Construct isolation forest. */
+public class IForest implements Serializable {
+ public final int numTrees;
+ public List<ITree> iTreeList;
+ public Double center0;
+ public Double center1;
+ public int subSamplesSize;
+
+ public IForest(int numTrees) {
+ this.numTrees = numTrees;
+ this.iTreeList = new ArrayList<>(256);
+ this.center0 = null;
+ this.center1 = null;
+ }
+
+ public void generateIsolationForest(DenseVector[] samplesData, int[] featureIndices) {
+ int n = samplesData.length;
+ this.subSamplesSize = Math.min(256, n);
Review Comment:
OK, I will fix it.
--
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