This is an automated email from the ASF dual-hosted git repository.
lindong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new 981d36a [FLINK-27742] Fix compatibility issue between stages
981d36a is described below
commit 981d36a9ef9c68464184c525de2c9bfd48fdc4e1
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Mon May 23 19:46:46 2022 +0800
[FLINK-27742] Fix compatibility issue between stages
This closes #103.
---
.../org/apache/flink/ml/linalg/DenseMatrix.java | 5 +
.../org/apache/flink/ml/linalg/DenseVector.java | 29 +++++
.../java/org/apache/flink/ml/linalg/Matrix.java | 3 +
.../org/apache/flink/ml/linalg/SparseVector.java | 10 ++
.../java/org/apache/flink/ml/linalg/Vector.java | 7 +-
.../apache/flink/ml/classification/knn/Knn.java | 5 +-
.../flink/ml/classification/knn/KnnModel.java | 3 +-
.../flink/ml/classification/knn/KnnModelData.java | 8 +-
.../ml/classification/linearsvc/LinearSVC.java | 11 +-
.../classification/linearsvc/LinearSVCModel.java | 3 +-
.../linearsvc/LinearSVCModelData.java | 3 +-
.../logisticregression/LogisticRegression.java | 11 +-
.../LogisticRegressionModel.java | 3 +-
.../LogisticRegressionModelData.java | 3 +-
.../naivebayes/NaiveBayesModelData.java | 5 +-
.../apache/flink/ml/clustering/kmeans/KMeans.java | 3 +-
.../flink/ml/clustering/kmeans/KMeansModel.java | 3 +-
.../ml/clustering/kmeans/KMeansModelData.java | 7 +-
.../flink/ml/clustering/kmeans/OnlineKMeans.java | 3 +-
.../ml/clustering/kmeans/OnlineKMeansModel.java | 3 +-
.../BinaryClassificationEvaluator.java | 11 +-
.../flink/ml/feature/bucketizer/Bucketizer.java | 6 +-
.../ml/feature/minmaxscaler/MinMaxScaler.java | 3 +-
.../ml/feature/minmaxscaler/MinMaxScalerModel.java | 3 +-
.../feature/stringindexer/StringIndexerModel.java | 12 +--
.../linearregression/LinearRegression.java | 11 +-
.../linearregression/LinearRegressionModel.java | 3 +-
.../apache/flink/ml/classification/KnnTest.java | 31 ++++--
.../flink/ml/classification/LinearSVCTest.java | 29 +++--
.../ml/classification/LogisticRegressionTest.java | 26 ++++-
.../flink/ml/classification/NaiveBayesTest.java | 54 ++++++----
.../org/apache/flink/ml/clustering/KMeansTest.java | 34 ++++--
.../flink/ml/clustering/OnlineKMeansTest.java | 53 ++++++++--
.../BinaryClassificationEvaluatorTest.java | 33 +++++-
.../apache/flink/ml/feature/BucketizerTest.java | 36 ++++++-
.../apache/flink/ml/feature/MinMaxScalerTest.java | 24 ++++-
.../apache/flink/ml/feature/OneHotEncoderTest.java | 22 +++-
.../flink/ml/feature/StandardScalerTest.java | 18 +++-
.../flink/ml/feature/VectorAssemblerTest.java | 25 ++++-
.../stringindexer/IndexToStringModelTest.java | 4 +-
.../feature/stringindexer/StringIndexerTest.java | 38 ++++---
.../flink/ml/regression/LinearRegressionTest.java | 22 +++-
.../org/apache/flink/ml/util/StageTestUtils.java | 52 ---------
.../java/org/apache/flink/ml/util/TestUtils.java | 117 +++++++++++++++++++++
44 files changed, 610 insertions(+), 185 deletions(-)
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseMatrix.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseMatrix.java
index 80c85b0..b37462d 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseMatrix.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseMatrix.java
@@ -82,4 +82,9 @@ public class DenseMatrix implements Matrix {
Preconditions.checkArgument(i >= 0 && i < numRows && j >= 0 && j <
numCols);
return values[numRows * j + i];
}
+
+ @Override
+ public DenseMatrix toDense() {
+ return this;
+ }
}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseVector.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseVector.java
index a4ccf9c..b4f0603 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseVector.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseVector.java
@@ -51,6 +51,35 @@ public class DenseVector implements Vector {
return values;
}
+ @Override
+ public DenseVector toDense() {
+ return this;
+ }
+
+ @Override
+ public SparseVector toSparse() {
+ int numNonZeros = 0;
+ for (double value : values) {
+ if (value != 0.0) {
+ numNonZeros++;
+ }
+ }
+
+ int[] nonZeroIndices = new int[numNonZeros];
+ double[] numZeroValues = new double[numNonZeros];
+ int k = 0;
+ for (int i = 0; i < values.length; i++) {
+ if (values[i] == 0.0) {
+ continue;
+ }
+ nonZeroIndices[k] = i;
+ numZeroValues[k] = values[i];
+ k++;
+ }
+
+ return new SparseVector(size(), nonZeroIndices, numZeroValues);
+ }
+
@Override
public String toString() {
return Arrays.toString(values);
diff --git a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Matrix.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Matrix.java
index d4e0897..d1bf1af 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Matrix.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Matrix.java
@@ -31,4 +31,7 @@ public interface Matrix extends Serializable {
/** Gets value of the (i,j) element. */
double get(int i, int j);
+
+ /** Converts the instance to a dense matrix. */
+ DenseMatrix toDense();
}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/SparseVector.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/SparseVector.java
index 6f7d128..f241e30 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/SparseVector.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/SparseVector.java
@@ -65,6 +65,16 @@ public class SparseVector implements Vector {
return result;
}
+ @Override
+ public DenseVector toDense() {
+ return new DenseVector(toArray());
+ }
+
+ @Override
+ public SparseVector toSparse() {
+ return this;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Vector.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Vector.java
index d5cce04..21718b3 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Vector.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Vector.java
@@ -33,9 +33,10 @@ public interface Vector extends Serializable {
double[] toArray();
/** Converts the instance to a dense vector. */
- default DenseVector toDense() {
- return new DenseVector(toArray());
- }
+ DenseVector toDense();
+
+ /** Converts the instance to a sparse vector. */
+ SparseVector toSparse();
/** Makes a deep copy of the vector. */
Vector clone();
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/Knn.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/Knn.java
index 93c0e5c..36e9d37 100644
--- a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/Knn.java
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/Knn.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.datastream.DataStreamUtils;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseMatrix;
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;
@@ -147,8 +148,8 @@ public class Knn implements Estimator<Knn, KnnModel>,
KnnParams<Knn> {
new MapFunction<Row, Tuple3<DenseVector, Double, Double>>() {
@Override
public Tuple3<DenseVector, Double, Double> map(Row value) {
- Double label = (Double) value.getField(getLabelCol());
- DenseVector feature = (DenseVector)
value.getField(getFeaturesCol());
+ Double label = ((Number)
value.getField(getLabelCol())).doubleValue();
+ DenseVector feature = ((Vector)
value.getField(getFeaturesCol())).toDense();
return Tuple3.of(feature, label,
Math.pow(BLAS.norm2(feature), 2));
}
});
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModel.java
index 28dcaea..3194fbb 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModel.java
@@ -27,6 +27,7 @@ import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.param.Param;
import org.apache.flink.ml.util.ParamUtils;
import org.apache.flink.ml.util.ReadWriteUtils;
@@ -145,7 +146,7 @@ public class KnnModel implements Model<KnnModel>,
KnnModelParams<KnnModel> {
getRuntimeContext().getBroadcastVariable(broadcastKey).get(0);
distanceVector = new DenseVector(knnModelData.labels.size());
}
- DenseVector feature = (DenseVector) row.getField(featureCol);
+ DenseVector feature = ((Vector)
row.getField(featureCol)).toDense();
double prediction = predictLabel(feature);
return Row.join(row, Row.of(prediction));
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModelData.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModelData.java
index 4bf0adb..d07859a 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModelData.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/knn/KnnModelData.java
@@ -29,6 +29,8 @@ import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.ml.linalg.DenseMatrix;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Matrix;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.typeinfo.DenseMatrixSerializer;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -74,9 +76,9 @@ public class KnnModelData {
.map(
x ->
new KnnModelData(
- (DenseMatrix) x.getField(0),
- (DenseVector) x.getField(1),
- (DenseVector) x.getField(2)));
+ ((Matrix) x.getField(0)).toDense(),
+ ((Vector) x.getField(1)).toDense(),
+ ((Vector) x.getField(2)).toDense()));
}
/** Encoder for {@link KnnModelData}. */
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVC.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVC.java
index 4169d48..8a8d1cb 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVC.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVC.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.lossfunc.HingeLoss;
import org.apache.flink.ml.common.optimizer.Optimizer;
import org.apache.flink.ml.common.optimizer.SGD;
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;
@@ -66,15 +67,19 @@ public class LinearSVC implements Estimator<LinearSVC,
LinearSVCModel>, LinearSV
double weight =
getWeightCol() == null
? 1.0
- : (Double)
dataPoint.getField(getWeightCol());
- double label = (Double)
dataPoint.getField(getLabelCol());
+ : ((Number)
dataPoint.getField(getWeightCol()))
+ .doubleValue();
+ double label =
+ ((Number)
dataPoint.getField(getLabelCol()))
+ .doubleValue();
Preconditions.checkState(
Double.compare(0.0, label) == 0
|| Double.compare(1.0,
label) == 0,
"LinearSVC only supports binary
classification. But detected label: %s.",
label);
DenseVector features =
- (DenseVector)
dataPoint.getField(getFeaturesCol());
+ ((Vector)
dataPoint.getField(getFeaturesCol()))
+ .toDense();
return new
LabeledPointWithWeight(features, label, weight);
});
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModel.java
index 253bbcc..a7da2e3 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModel.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.param.Param;
@@ -152,7 +153,7 @@ public class LinearSVCModel implements
Model<LinearSVCModel>, LinearSVCModelPara
getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
coefficient = modelData.coefficient;
}
- DenseVector features = (DenseVector)
dataPoint.getField(featuresCol);
+ DenseVector features = ((Vector)
dataPoint.getField(featuresCol)).toDense();
Row predictionResult = predictOneDataPoint(features, coefficient,
threshold);
return Row.join(dataPoint, predictionResult);
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModelData.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModelData.java
index 96e8a27..d181c6c 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModelData.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/linearsvc/LinearSVCModelData.java
@@ -26,6 +26,7 @@ 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.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
@@ -62,7 +63,7 @@ public class LinearSVCModelData {
StreamTableEnvironment tEnv =
(StreamTableEnvironment) ((TableImpl)
modelData).getTableEnvironment();
return tEnv.toDataStream(modelData)
- .map(x -> new LinearSVCModelData((DenseVector) x.getField(0)));
+ .map(x -> new LinearSVCModelData(((Vector)
x.getField(0)).toDense()));
}
/** Data encoder for {@link LinearSVCModel}. */
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegression.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegression.java
index f64fc10..df8c386 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegression.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegression.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.lossfunc.BinaryLogisticLoss;
import org.apache.flink.ml.common.optimizer.Optimizer;
import org.apache.flink.ml.common.optimizer.SGD;
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;
@@ -72,8 +73,11 @@ public class LogisticRegression
double weight =
getWeightCol() == null
? 1.0
- : (Double)
dataPoint.getField(getWeightCol());
- double label = (Double)
dataPoint.getField(getLabelCol());
+ : ((Number)
dataPoint.getField(getWeightCol()))
+ .doubleValue();
+ double label =
+ ((Number)
dataPoint.getField(getLabelCol()))
+ .doubleValue();
boolean isBinomial =
Double.compare(0., label) == 0
|| Double.compare(1.,
label) == 0;
@@ -82,7 +86,8 @@ public class LogisticRegression
"Multinomial classification is
not supported yet. Supported options: [auto, binomial].");
}
DenseVector features =
- (DenseVector)
dataPoint.getField(getFeaturesCol());
+ ((Vector)
dataPoint.getField(getFeaturesCol()))
+ .toDense();
return new
LabeledPointWithWeight(features, label, weight);
});
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModel.java
index 247a4d5..ac42142 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModel.java
@@ -27,6 +27,7 @@ import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.param.Param;
import org.apache.flink.ml.util.ParamUtils;
@@ -148,7 +149,7 @@ public class LogisticRegressionModel
getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
coefficient = modelData.coefficient;
}
- DenseVector features = (DenseVector)
dataPoint.getField(featuresCol);
+ DenseVector features = ((Vector)
dataPoint.getField(featuresCol)).toDense();
Row predictionResult = predictOneDataPoint(features, coefficient);
return Row.join(dataPoint, predictionResult);
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModelData.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModelData.java
index 774c19e..d2f451f 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModelData.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/LogisticRegressionModelData.java
@@ -26,6 +26,7 @@ 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.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
@@ -62,7 +63,7 @@ public class LogisticRegressionModelData {
StreamTableEnvironment tEnv =
(StreamTableEnvironment) ((TableImpl)
modelData).getTableEnvironment();
return tEnv.toDataStream(modelData)
- .map(x -> new LogisticRegressionModelData((DenseVector)
x.getField(0)));
+ .map(x -> new LogisticRegressionModelData(((Vector)
x.getField(0)).toDense()));
}
/** Data encoder for {@link LogisticRegressionModel}. */
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/naivebayes/NaiveBayesModelData.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/naivebayes/NaiveBayesModelData.java
index a03141d..80ff835 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/naivebayes/NaiveBayesModelData.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/classification/naivebayes/NaiveBayesModelData.java
@@ -29,6 +29,7 @@ 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.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
@@ -85,8 +86,8 @@ public class NaiveBayesModelData {
row ->
new NaiveBayesModelData(
(Map<Double, Double>[][])
row.getField(0),
- (DenseVector) row.getField(1),
- (DenseVector)
row.getField(2)));
+ ((Vector)
row.getField(1)).toDense(),
+ ((Vector)
row.getField(2)).toDense()));
}
/** Data encoder for the {@link NaiveBayesModelData}. */
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeans.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeans.java
index 31e6d6c..2aa2c3c 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeans.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeans.java
@@ -44,6 +44,7 @@ import org.apache.flink.ml.common.distance.DistanceMeasure;
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;
@@ -92,7 +93,7 @@ public class KMeans implements Estimator<KMeans,
KMeansModel>, KMeansParams<KMea
(StreamTableEnvironment) ((TableImpl)
inputs[0]).getTableEnvironment();
DataStream<DenseVector> points =
tEnv.toDataStream(inputs[0])
- .map(row -> (DenseVector)
row.getField(getFeaturesCol()));
+ .map(row -> ((Vector)
row.getField(getFeaturesCol())).toDense());
DataStream<DenseVector[]> initCentroids =
selectRandomCentroids(points, getK(), getSeed());
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModel.java
index d569d97..9cfea62 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModel.java
@@ -27,6 +27,7 @@ import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.common.distance.DistanceMeasure;
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;
@@ -132,7 +133,7 @@ public class KMeansModel implements Model<KMeansModel>,
KMeansModelParams<KMeans
Preconditions.checkArgument(modelData.centroids.length <= k);
centroids = modelData.centroids;
}
- DenseVector point = (DenseVector) dataPoint.getField(featuresCol);
+ DenseVector point = ((Vector)
dataPoint.getField(featuresCol)).toDense();
int closestCentroidId = KMeans.findClosestCentroidId(centroids,
point, distanceMeasure);
return Row.join(dataPoint, Row.of(closestCentroidId));
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModelData.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModelData.java
index a7a62c7..34c69ac 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModelData.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeansModelData.java
@@ -29,6 +29,7 @@ import
org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
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.linalg.typeinfo.DenseVectorSerializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -129,8 +130,10 @@ public class KMeansModelData {
.map(
x ->
new KMeansModelData(
- (DenseVector[]) x.getField(0),
- (DenseVector) x.getField(1)));
+ Arrays.stream(((Vector[])
x.getField(0)))
+ .map(Vector::toDense)
+ .toArray(DenseVector[]::new),
+ ((Vector) x.getField(1)).toDense()));
}
/** Data encoder for {@link KMeansModelData}. */
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeans.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeans.java
index 7ede3e4..b876b13 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeans.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeans.java
@@ -34,6 +34,7 @@ import org.apache.flink.ml.api.Estimator;
import org.apache.flink.ml.common.distance.DistanceMeasure;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.param.Param;
import org.apache.flink.ml.util.ParamUtils;
@@ -335,7 +336,7 @@ public class OnlineKMeans
@Override
public DenseVector map(Row row) {
- return (DenseVector) row.getField(featuresCol);
+ return ((Vector) row.getField(featuresCol)).toDense();
}
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeansModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeansModel.java
index 3643d19..6a7c852 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeansModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/OnlineKMeansModel.java
@@ -27,6 +27,7 @@ import org.apache.flink.ml.api.Model;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.common.distance.DistanceMeasure;
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;
@@ -171,7 +172,7 @@ public class OnlineKMeansModel
bufferedPointsState.add(dataPoint);
return;
}
- DenseVector point = (DenseVector) dataPoint.getField(featuresCol);
+ DenseVector point = ((Vector)
dataPoint.getField(featuresCol)).toDense();
int closestCentroidId = KMeans.findClosestCentroidId(centroids,
point, distanceMeasure);
output.collect(new StreamRecord<>(Row.join(dataPoint,
Row.of(closestCentroidId))));
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java
index 49cf4ae..a80fc7e 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java
@@ -35,7 +35,7 @@ import org.apache.flink.iteration.operator.OperatorStateUtils;
import org.apache.flink.ml.api.AlgoOperator;
import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.DataStreamUtils;
-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;
@@ -658,10 +658,11 @@ public class BinaryClassificationEvaluator
double label = ((Number) value.getFieldAs(labelCol)).doubleValue();
Object probOrigin = value.getField(rawPredictionCol);
double prob =
- probOrigin instanceof DenseVector
- ? ((DenseVector) probOrigin).get(1)
- : (double) probOrigin;
- double weight = weightCol == null ? 1.0 :
value.getFieldAs(weightCol);
+ probOrigin instanceof Vector
+ ? ((Vector) probOrigin).get(1)
+ : ((Number) probOrigin).doubleValue();
+ double weight =
+ weightCol == null ? 1.0 : ((Number)
value.getField(weightCol)).doubleValue();
return Tuple3.of(prob, label == 1.0, weight);
}
}
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/bucketizer/Bucketizer.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/bucketizer/Bucketizer.java
index 8afefcb..1382963 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/bucketizer/Bucketizer.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/bucketizer/Bucketizer.java
@@ -71,7 +71,7 @@ public class Bucketizer implements Transformer<Bucketizer>,
BucketizerParams<Buc
RowTypeInfo inputTypeInfo =
TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
TypeInformation<?>[] outputTypes = new
TypeInformation[outputCols.length];
- Arrays.fill(outputTypes, BasicTypeInfo.INT_TYPE_INFO);
+ Arrays.fill(outputTypes, BasicTypeInfo.DOUBLE_TYPE_INFO);
RowTypeInfo outputTypeInfo =
new RowTypeInfo(
ArrayUtils.addAll(inputTypeInfo.getFieldTypes(),
outputTypes),
@@ -108,7 +108,7 @@ public class Bucketizer implements Transformer<Bucketizer>,
BucketizerParams<Buc
boolean isInvalid = false;
if (!Double.isNaN(feature)) {
- int index = Arrays.binarySearch(splits, feature);
+ double index = Arrays.binarySearch(splits, feature);
if (index >= 0) {
if (index == splits.length - 1) {
index--;
@@ -136,7 +136,7 @@ public class Bucketizer implements Transformer<Bucketizer>,
BucketizerParams<Buc
case SKIP_INVALID:
return;
case KEEP_INVALID:
- outputRow.setField(i, splits.length - 1);
+ outputRow.setField(i, (double) splits.length - 1);
break;
default:
throw new UnsupportedOperationException(
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScaler.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScaler.java
index 43c5ac9..0ba885b 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScaler.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScaler.java
@@ -27,6 +27,7 @@ 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.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;
@@ -81,7 +82,7 @@ public class MinMaxScaler
tEnv.toDataStream(inputs[0])
.map(
(MapFunction<Row, DenseVector>)
- value -> (DenseVector)
value.getField(inputCol));
+ value -> ((Vector)
value.getField(inputCol)).toDense());
DataStream<DenseVector> minMaxValues =
inputData
.transform(
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScalerModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScalerModel.java
index 87b5993..59b6aa0 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScalerModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScalerModel.java
@@ -25,6 +25,7 @@ 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;
@@ -166,7 +167,7 @@ public class MinMaxScalerModel
}
}
}
- DenseVector inputVec = (DenseVector) row.getField(inputCol);
+ DenseVector inputVec = ((Vector) row.getField(inputCol)).toDense();
DenseVector outputVec = new DenseVector(scaleVector.size());
for (int i = 0; i < scaleVector.size(); ++i) {
outputVec.values[i] =
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stringindexer/StringIndexerModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stringindexer/StringIndexerModel.java
index f1257cc..4781966 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stringindexer/StringIndexerModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stringindexer/StringIndexerModel.java
@@ -46,7 +46,7 @@ import java.util.HashMap;
import java.util.Map;
/**
- * A Model which transforms input string/numeric column(s) to integer
column(s) using the model data
+ * A Model which transforms input string/numeric column(s) to double column(s)
using the model data
* computed by {@link StringIndexer}.
*
* <p>The `keep` option of {@link HasHandleInvalid} means that we put the
invalid entries in a
@@ -107,7 +107,7 @@ public class StringIndexerModel
RowTypeInfo inputTypeInfo =
TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
TypeInformation<?>[] outputTypes = new
TypeInformation[outputCols.length];
- Arrays.fill(outputTypes, BasicTypeInfo.INT_TYPE_INFO);
+ Arrays.fill(outputTypes, BasicTypeInfo.DOUBLE_TYPE_INFO);
RowTypeInfo outputTypeInfo =
new RowTypeInfo(
ArrayUtils.addAll(inputTypeInfo.getFieldTypes(),
outputTypes),
@@ -132,9 +132,9 @@ public class StringIndexerModel
return new Table[] {tEnv.fromDataStream(result)};
}
- /** Maps the input columns to integer values according to the model data.
*/
+ /** Maps the input columns to double values according to the model data. */
private static class String2Index extends RichFlatMapFunction<Row, Row> {
- private HashMap<String, Integer>[] modelDataMap;
+ private HashMap<String, Double>[] modelDataMap;
private final String broadcastModelKey;
private final String[] inputCols;
private final String handleInValid;
@@ -155,7 +155,7 @@ public class StringIndexerModel
getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
String[][] stringsArray = modelData.stringArrays;
for (int i = 0; i < stringsArray.length; i++) {
- int idx = 0;
+ double idx = 0.0;
modelDataMap[i] = new HashMap<>(stringsArray[i].length);
for (String string : stringsArray[i]) {
modelDataMap[i].put(string, idx++);
@@ -190,7 +190,7 @@ public class StringIndexerModel
+ HANDLE_INVALID
+ " parameter for more options.");
case KEEP_INVALID:
- outputIndices.setField(i, modelDataMap[i].size());
+ outputIndices.setField(i, (double)
modelDataMap[i].size());
break;
default:
throw new UnsupportedOperationException(
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegression.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegression.java
index 7edffc4..3de4804 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegression.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegression.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.lossfunc.LeastSquareLoss;
import org.apache.flink.ml.common.optimizer.Optimizer;
import org.apache.flink.ml.common.optimizer.SGD;
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;
@@ -67,10 +68,14 @@ public class LinearRegression
double weight =
getWeightCol() == null
? 1.0
- : (Double)
dataPoint.getField(getWeightCol());
- double label = (Double)
dataPoint.getField(getLabelCol());
+ : ((Number)
dataPoint.getField(getWeightCol()))
+ .doubleValue();
+ double label =
+ ((Number)
dataPoint.getField(getLabelCol()))
+ .doubleValue();
DenseVector features =
- (DenseVector)
dataPoint.getField(getFeaturesCol());
+ ((Vector)
dataPoint.getField(getFeaturesCol()))
+ .toDense();
return new
LabeledPointWithWeight(features, label, weight);
});
diff --git
a/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegressionModel.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegressionModel.java
index 7fb9019..dec2f43 100644
---
a/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegressionModel.java
+++
b/flink-ml-lib/src/main/java/org/apache/flink/ml/regression/linearregression/LinearRegressionModel.java
@@ -26,6 +26,7 @@ import org.apache.flink.ml.common.broadcast.BroadcastUtils;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.linalg.BLAS;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.param.Param;
import org.apache.flink.ml.util.ParamUtils;
import org.apache.flink.ml.util.ReadWriteUtils;
@@ -141,7 +142,7 @@ public class LinearRegressionModel
getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
coefficient = modelData.coefficient;
}
- DenseVector features = (DenseVector)
dataPoint.getField(featuresCol);
+ DenseVector features = ((Vector)
dataPoint.getField(featuresCol)).toDense();
Row predictionResult = predictOneDataPoint(features, coefficient);
return Row.join(dataPoint, predictionResult);
}
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/KnnTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/KnnTest.java
index e395951..6121725 100644
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/KnnTest.java
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/KnnTest.java
@@ -27,9 +27,10 @@ import org.apache.flink.ml.classification.knn.KnnModel;
import org.apache.flink.ml.classification.knn.KnnModelData;
import org.apache.flink.ml.linalg.DenseMatrix;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -51,6 +52,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
/** Tests {@link Knn} and {@link KnnModel}. */
@@ -121,7 +123,7 @@ public class KnnTest {
@Override
public Tuple2<Double, Double> map(Row row)
{
return Tuple2.of(
- (Double)
row.getField(labelCol),
+ ((Number)
row.getField(labelCol)).doubleValue(),
(Double)
row.getField(predictionCol));
}
});
@@ -179,15 +181,31 @@ public class KnnTest {
verifyPredictionResult(output, knn.getLabelCol(),
knn.getPredictionCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainData = TestUtils.convertDataTypesToSparseInt(tEnv, trainData);
+ predictData = TestUtils.convertDataTypesToSparseInt(tEnv, predictData);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class},
+ TestUtils.getColumnDataTypes(trainData));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class},
+ TestUtils.getColumnDataTypes(predictData));
+
+ Knn knn = new Knn();
+ KnnModel knnModel = knn.fit(trainData);
+ Table output = knnModel.transform(predictData)[0];
+ verifyPredictionResult(output, knn.getLabelCol(),
knn.getPredictionCol());
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
Knn knn = new Knn();
Knn loadedKnn =
- StageTestUtils.saveAndReload(tEnv, knn,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, knn,
tempFolder.newFolder().getAbsolutePath());
KnnModel knnModel = loadedKnn.fit(trainData);
knnModel =
- StageTestUtils.saveAndReload(
- tEnv, knnModel,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, knnModel,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Arrays.asList("packedFeatures", "featureNormSquares",
"labels"),
knnModel.getModelData()[0].getResolvedSchema().getColumnNames());
@@ -200,8 +218,7 @@ public class KnnTest {
Knn knn = new Knn();
KnnModel knnModel = knn.fit(trainData);
KnnModel newModel =
- StageTestUtils.saveAndReload(
- tEnv, knnModel,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, knnModel,
tempFolder.newFolder().getAbsolutePath());
Table output = newModel.transform(predictData)[0];
verifyPredictionResult(output, knn.getLabelCol(),
knn.getPredictionCol());
}
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LinearSVCTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LinearSVCTest.java
index 156244e..2c03919 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LinearSVCTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LinearSVCTest.java
@@ -27,10 +27,12 @@ import
org.apache.flink.ml.classification.linearsvc.LinearSVC;
import org.apache.flink.ml.classification.linearsvc.LinearSVCModel;
import org.apache.flink.ml.classification.linearsvc.LinearSVCModelData;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -109,8 +111,8 @@ public class LinearSVCTest {
throws Exception {
List<Row> predResult =
IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
for (Row predictionRow : predResult) {
- DenseVector feature = (DenseVector)
predictionRow.getField(featuresCol);
- double prediction = (double) predictionRow.getField(predictionCol);
+ DenseVector feature = ((Vector)
predictionRow.getField(featuresCol)).toDense();
+ double prediction = (Double) predictionRow.getField(predictionCol);
DenseVector rawPrediction = (DenseVector)
predictionRow.getField(rawPredictionCol);
if (feature.get(0) <= 5) {
assertEquals(0, prediction, TOLERANCE);
@@ -197,14 +199,29 @@ public class LinearSVCTest {
linearSVC.getRawPredictionCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
trainDataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class,
Integer.class},
+ TestUtils.getColumnDataTypes(trainDataTable));
+
+ LinearSVC linearSVC = new LinearSVC().setWeightCol("weight");
+ Table output =
linearSVC.fit(trainDataTable).transform(trainDataTable)[0];
+ verifyPredictionResult(
+ output,
+ linearSVC.getFeaturesCol(),
+ linearSVC.getPredictionCol(),
+ linearSVC.getRawPredictionCol());
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
LinearSVC linearSVC = new LinearSVC().setWeightCol("weight");
linearSVC =
- StageTestUtils.saveAndReload(
- tEnv, linearSVC,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, linearSVC,
tempFolder.newFolder().getAbsolutePath());
LinearSVCModel model = linearSVC.fit(trainDataTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Collections.singletonList("coefficient"),
model.getModelData()[0].getResolvedSchema().getColumnNames());
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LogisticRegressionTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LogisticRegressionTest.java
index 4bdab6b..129fd2e 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LogisticRegressionTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/LogisticRegressionTest.java
@@ -27,10 +27,12 @@ import
org.apache.flink.ml.classification.logisticregression.LogisticRegression;
import
org.apache.flink.ml.classification.logisticregression.LogisticRegressionModel;
import
org.apache.flink.ml.classification.logisticregression.LogisticRegressionModelData;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -135,7 +137,7 @@ public class LogisticRegressionTest {
throws Exception {
List<Row> predResult =
IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
for (Row predictionRow : predResult) {
- DenseVector feature = (DenseVector)
predictionRow.getField(featuresCol);
+ DenseVector feature = ((Vector)
predictionRow.getField(featuresCol)).toDense();
double prediction = (double) predictionRow.getField(predictionCol);
DenseVector rawPrediction = (DenseVector)
predictionRow.getField(rawPredictionCol);
if (feature.get(0) <= 5) {
@@ -223,14 +225,30 @@ public class LogisticRegressionTest {
logisticRegression.getRawPredictionCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ binomialDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
binomialDataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class,
Integer.class},
+ TestUtils.getColumnDataTypes(binomialDataTable));
+
+ LogisticRegression logisticRegression = new
LogisticRegression().setWeightCol("weight");
+ Table output =
logisticRegression.fit(binomialDataTable).transform(binomialDataTable)[0];
+ verifyPredictionResult(
+ output,
+ logisticRegression.getFeaturesCol(),
+ logisticRegression.getPredictionCol(),
+ logisticRegression.getRawPredictionCol());
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
LogisticRegression logisticRegression = new
LogisticRegression().setWeightCol("weight");
logisticRegression =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
tEnv, logisticRegression,
tempFolder.newFolder().getAbsolutePath());
LogisticRegressionModel model =
logisticRegression.fit(binomialDataTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Collections.singletonList("coefficient"),
model.getModelData()[0].getResolvedSchema().getColumnNames());
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/NaiveBayesTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/NaiveBayesTest.java
index c481fbb..bffb40a 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/NaiveBayesTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/classification/NaiveBayesTest.java
@@ -23,10 +23,11 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.ml.classification.naivebayes.NaiveBayes;
import org.apache.flink.ml.classification.naivebayes.NaiveBayesModel;
import org.apache.flink.ml.classification.naivebayes.NaiveBayesModelData;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -57,7 +58,7 @@ public class NaiveBayesTest {
private StreamTableEnvironment tEnv;
private Table trainTable;
private Table predictTable;
- private Map<Vector, Integer> expectedOutput;
+ private Map<Vector, Double> expectedOutput;
private NaiveBayes estimator;
@Before
@@ -88,12 +89,12 @@ public class NaiveBayesTest {
predictTable =
tEnv.fromDataStream(env.fromCollection(predictData)).as("features");
expectedOutput =
- new HashMap<Vector, Integer>() {
+ new HashMap<Vector, Double>() {
{
- put(Vectors.dense(0, 1.), 11);
- put(Vectors.dense(0, 0.), 11);
- put(Vectors.dense(1, 0.), 10);
- put(Vectors.dense(1, 1.), 10);
+ put(Vectors.dense(0, 1.), 11.0);
+ put(Vectors.dense(0, 0.), 11.0);
+ put(Vectors.dense(1, 0.), 10.0);
+ put(Vectors.dense(1, 1.), 10.0);
}
};
@@ -115,14 +116,14 @@ public class NaiveBayesTest {
* @param predictionCol Name of the column in the table that contains the
prediction result
* @return A map containing the collected results
*/
- private static Map<Vector, Integer> executeAndCollect(
+ private static Map<Vector, Double> executeAndCollect(
Table table, String featuresCol, String predictionCol) {
- Map<Vector, Integer> map = new HashMap<>();
+ Map<Vector, Double> map = new HashMap<>();
for (CloseableIterator<Row> it = table.execute().collect();
it.hasNext(); ) {
Row row = it.next();
map.put(
- (Vector) row.getField(featuresCol),
- ((Number) row.getField(predictionCol)).intValue());
+ ((Vector) row.getField(featuresCol)).toDense(),
+ (Double) row.getField(predictionCol));
}
return map;
@@ -165,7 +166,25 @@ public class NaiveBayesTest {
public void testFitAndPredict() {
NaiveBayesModel model = estimator.fit(trainTable);
Table outputTable = model.transform(predictTable)[0];
- Map<Vector, Integer> actualOutput =
+ Map<Vector, Double> actualOutput =
+ executeAndCollect(outputTable, model.getFeaturesCol(),
model.getPredictionCol());
+ assertEquals(expectedOutput, actualOutput);
+ }
+
+ @Test
+ public void testInputTypeConversion() {
+ trainTable = TestUtils.convertDataTypesToSparseInt(tEnv, trainTable);
+ predictTable = TestUtils.convertDataTypesToSparseInt(tEnv,
predictTable);
+
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class},
+ TestUtils.getColumnDataTypes(trainTable));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
TestUtils.getColumnDataTypes(predictTable));
+
+ NaiveBayesModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ Map<Vector, Double> actualOutput =
executeAndCollect(outputTable, model.getFeaturesCol(),
model.getPredictionCol());
assertEquals(expectedOutput, actualOutput);
}
@@ -182,7 +201,7 @@ public class NaiveBayesTest {
NaiveBayesModel model = estimator.fit(trainTable);
Table outputTable = model.transform(predictTable)[0];
- Map<Vector, Integer> actualOutput =
+ Map<Vector, Double> actualOutput =
executeAndCollect(outputTable, model.getFeaturesCol(),
model.getPredictionCol());
assertEquals(expectedOutput, actualOutput);
}
@@ -253,16 +272,15 @@ public class NaiveBayesTest {
@Test
public void testSaveLoad() throws Exception {
estimator =
- StageTestUtils.saveAndReload(
- tEnv, estimator,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, estimator,
tempFolder.newFolder().getAbsolutePath());
NaiveBayesModel model = estimator.fit(trainTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
Table outputTable = model.transform(predictTable)[0];
- Map<Vector, Integer> actualOutput =
+ Map<Vector, Double> actualOutput =
executeAndCollect(outputTable, model.getFeaturesCol(),
model.getPredictionCol());
assertEquals(expectedOutput, actualOutput);
}
@@ -299,7 +317,7 @@ public class NaiveBayesTest {
Table outputTable = modelB.transform(predictTable)[0];
- Map<Vector, Integer> actualOutput =
+ Map<Vector, Double> actualOutput =
executeAndCollect(outputTable, modelB.getFeaturesCol(),
modelB.getPredictionCol());
assertEquals(expectedOutput, actualOutput);
}
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/KMeansTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/KMeansTest.java
index ac2cdb4..d1fb1c3 100644
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/KMeansTest.java
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/KMeansTest.java
@@ -25,9 +25,11 @@ import org.apache.flink.ml.clustering.kmeans.KMeansModel;
import org.apache.flink.ml.clustering.kmeans.KMeansModelData;
import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -97,8 +99,7 @@ public class KMeansTest extends AbstractTestBase {
env.setRestartStrategy(RestartStrategies.noRestart());
tEnv = StreamTableEnvironment.create(env);
- Schema schema = Schema.newBuilder().column("f0",
DataTypes.of(DenseVector.class)).build();
- dataTable = tEnv.fromDataStream(env.fromCollection(DATA),
schema).as("features");
+ dataTable =
tEnv.fromDataStream(env.fromCollection(DATA)).as("features");
}
/**
@@ -114,7 +115,7 @@ public class KMeansTest extends AbstractTestBase {
List<Row> rows, String featuresCol, String predictionCol) {
Map<Integer, Set<DenseVector>> map = new HashMap<>();
for (Row row : rows) {
- DenseVector vector = (DenseVector) row.getField(featuresCol);
+ DenseVector vector = ((Vector)
row.getField(featuresCol)).toDense();
int predict = (Integer) row.getField(predictionCol);
map.putIfAbsent(predict, new HashSet<>());
map.get(predict).add(vector);
@@ -203,15 +204,34 @@ public class KMeansTest extends AbstractTestBase {
assertTrue(CollectionUtils.isEqualCollection(expectedGroups,
actualGroups));
}
+ @Test
+ public void testInputTypeConversion() {
+ dataTable = TestUtils.convertDataTypesToSparseInt(tEnv, dataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
TestUtils.getColumnDataTypes(dataTable));
+
+ KMeans kmeans = new KMeans().setMaxIter(2).setK(2);
+ KMeansModel model = kmeans.fit(dataTable);
+ Table output = model.transform(dataTable)[0];
+
+ assertEquals(
+ Arrays.asList("features", "prediction"),
+ output.getResolvedSchema().getColumnNames());
+ List<Row> results = IteratorUtils.toList(output.execute().collect());
+ List<Set<DenseVector>> actualGroups =
+ groupFeaturesByPrediction(
+ results, kmeans.getFeaturesCol(),
kmeans.getPredictionCol());
+ assertTrue(CollectionUtils.isEqualCollection(expectedGroups,
actualGroups));
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
KMeans kmeans = new KMeans().setMaxIter(2).setK(2);
KMeans loadedKmeans =
- StageTestUtils.saveAndReload(
- tEnv, kmeans,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, kmeans,
tempFolder.newFolder().getAbsolutePath());
KMeansModel model = loadedKmeans.fit(dataTable);
KMeansModel loadedModel =
- StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
Table output = loadedModel.transform(dataTable)[0];
assertEquals(
Arrays.asList("centroids", "weights"),
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java
index cd4af68..b56528f 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java
@@ -32,10 +32,12 @@ import org.apache.flink.ml.clustering.kmeans.OnlineKMeans;
import org.apache.flink.ml.clustering.kmeans.OnlineKMeansModel;
import org.apache.flink.ml.common.distance.EuclideanDistanceMeasure;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.util.InMemorySinkFunction;
import org.apache.flink.ml.util.InMemorySourceFunction;
+import org.apache.flink.ml.util.TestUtils;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
@@ -66,6 +68,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static
org.apache.flink.ml.clustering.KMeansTest.groupFeaturesByPrediction;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
/** Tests {@link OnlineKMeans} and {@link OnlineKMeansModel}. */
@@ -307,6 +310,46 @@ public class OnlineKMeansTest extends TestLogger {
expectedGroups2, onlineKMeans.getFeaturesCol(),
onlineKMeans.getPredictionCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ offlineTrainTable = TestUtils.convertDataTypesToSparseInt(tEnv,
offlineTrainTable);
+ onlineTrainTable = TestUtils.convertDataTypesToSparseInt(tEnv,
onlineTrainTable);
+ onlinePredictTable = TestUtils.convertDataTypesToSparseInt(tEnv,
onlinePredictTable);
+
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
+ TestUtils.getColumnDataTypes(offlineTrainTable));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
+ TestUtils.getColumnDataTypes(onlineTrainTable));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
+ TestUtils.getColumnDataTypes(onlinePredictTable));
+
+ OnlineKMeans onlineKMeans =
+ new OnlineKMeans()
+ .setFeaturesCol("features")
+ .setPredictionCol("prediction")
+ .setGlobalBatchSize(6)
+ .setInitialModelData(
+ KMeansModelData.generateRandomModelData(tEnv,
2, 2, 0.0, 0));
+ OnlineKMeansModel onlineModel = onlineKMeans.fit(onlineTrainTable);
+ transformAndOutputData(onlineModel);
+
+ final JobID jobID = submitJob(env.getStreamGraph().getJobGraph());
+ waitInitModelDataSetup(jobID);
+
+ trainSource.addAll(trainData1);
+ waitModelDataUpdate(jobID);
+ predictAndAssert(
+ expectedGroups1, onlineKMeans.getFeaturesCol(),
onlineKMeans.getPredictionCol());
+
+ trainSource.addAll(trainData2);
+ waitModelDataUpdate(jobID);
+ predictAndAssert(
+ expectedGroups2, onlineKMeans.getFeaturesCol(),
onlineKMeans.getPredictionCol());
+ }
+
@Test
public void testInitWithKMeans() throws Exception {
KMeans kMeans = new
KMeans().setFeaturesCol("features").setPredictionCol("prediction");
@@ -361,12 +404,11 @@ public class OnlineKMeansTest extends TestLogger {
},
Vectors.dense(4.5, 4.5));
- Assert.assertArrayEquals(
- expectedModelData.weights.values,
actualModelData.weights.values, 1e-5);
+ assertArrayEquals(expectedModelData.weights.values,
actualModelData.weights.values, 1e-5);
Assert.assertEquals(expectedModelData.centroids.length,
actualModelData.centroids.length);
Arrays.sort(actualModelData.centroids,
Comparator.comparingDouble(vector -> vector.get(0)));
for (int i = 0; i < expectedModelData.centroids.length; i++) {
- Assert.assertArrayEquals(
+ assertArrayEquals(
expectedModelData.centroids[i].values,
actualModelData.centroids[i].values,
1e-5);
@@ -456,12 +498,11 @@ public class OnlineKMeansTest extends TestLogger {
new DenseVector[] {Vectors.dense(-10.2, 0.2),
Vectors.dense(10.1, 0.1)},
Vectors.dense(3, 3));
- Assert.assertArrayEquals(
- expectedModelData.weights.values,
actualModelData.weights.values, 1e-5);
+ assertArrayEquals(expectedModelData.weights.values,
actualModelData.weights.values, 1e-5);
Assert.assertEquals(expectedModelData.centroids.length,
actualModelData.centroids.length);
Arrays.sort(actualModelData.centroids,
Comparator.comparingDouble(vector -> vector.get(0)));
for (int i = 0; i < expectedModelData.centroids.length; i++) {
- Assert.assertArrayEquals(
+ assertArrayEquals(
expectedModelData.centroids[i].values,
actualModelData.centroids[i].values,
1e-5);
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/evaluation/BinaryClassificationEvaluatorTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/evaluation/BinaryClassificationEvaluatorTest.java
index e40502f..787a024 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/evaluation/BinaryClassificationEvaluatorTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/evaluation/BinaryClassificationEvaluatorTest.java
@@ -22,8 +22,9 @@ import
org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import
org.apache.flink.ml.evaluation.binaryclassification.BinaryClassificationEvaluator;
import
org.apache.flink.ml.evaluation.binaryclassification.BinaryClassificationEvaluatorParams;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -193,6 +194,34 @@ public class BinaryClassificationEvaluatorTest {
}
}
+ @Test
+ public void testInputTypeConversion() {
+ inputDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
inputDataTable);
+ assertArrayEquals(
+ new Class<?>[] {Integer.class, SparseVector.class},
+ TestUtils.getColumnDataTypes(inputDataTable));
+
+ BinaryClassificationEvaluator eval =
+ new BinaryClassificationEvaluator()
+ .setMetricsNames(
+
BinaryClassificationEvaluatorParams.AREA_UNDER_PR,
+ BinaryClassificationEvaluatorParams.KS,
+
BinaryClassificationEvaluatorParams.AREA_UNDER_ROC);
+ Table evalResult = eval.transform(inputDataTable)[0];
+ List<Row> results =
IteratorUtils.toList(evalResult.execute().collect());
+ assertArrayEquals(
+ new String[] {
+ BinaryClassificationEvaluatorParams.AREA_UNDER_PR,
+ BinaryClassificationEvaluatorParams.KS,
+ BinaryClassificationEvaluatorParams.AREA_UNDER_ROC
+ },
+ evalResult.getResolvedSchema().getColumnNames().toArray());
+ Row result = results.get(0);
+ for (int i = 0; i < EXPECTED_DATA.length; ++i) {
+ assertEquals(EXPECTED_DATA[i], result.getFieldAs(i), EPS);
+ }
+ }
+
@Test
public void testEvaluateWithDoubleRaw() {
BinaryClassificationEvaluator eval =
@@ -288,7 +317,7 @@ public class BinaryClassificationEvaluatorTest {
BinaryClassificationEvaluatorParams.KS,
BinaryClassificationEvaluatorParams.AREA_UNDER_ROC);
BinaryClassificationEvaluator loadedEval =
- StageTestUtils.saveAndReload(tEnv, eval,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, eval,
tempFolder.newFolder().getAbsolutePath());
Table evalResult = loadedEval.transform(inputDataTable)[0];
assertArrayEquals(
new String[] {
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/BucketizerTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/BucketizerTest.java
index 7170103..6f64a2e 100644
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/BucketizerTest.java
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/BucketizerTest.java
@@ -22,7 +22,7 @@ import
org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.ml.common.param.HasHandleInvalid;
import org.apache.flink.ml.feature.bucketizer.Bucketizer;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -68,9 +68,13 @@ public class BucketizerTest extends AbstractTestBase {
};
private final List<Row> expectedKeepResult =
- Arrays.asList(Row.of(1, 0, 1, 0, 0), Row.of(2, 2, 1, 1, 0),
Row.of(3, 2, 0, 0, 1));
+ Arrays.asList(
+ Row.of(1, 0.0, 1.0, 0.0, 0.0),
+ Row.of(2, 2.0, 1.0, 1.0, 0.0),
+ Row.of(3, 2.0, 0.0, 0.0, 1.0));
- private final List<Row> expectedSkipResult =
Collections.singletonList(Row.of(1, 0, 1, 0, 0));
+ private final List<Row> expectedSkipResult =
+ Collections.singletonList(Row.of(1, 0.0, 1.0, 0.0, 0.0));
@Before
public void before() {
@@ -175,6 +179,29 @@ public class BucketizerTest extends AbstractTestBase {
}
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ inputTable = TestUtils.convertDataTypesToSparseInt(tEnv, inputTable);
+ assertArrayEquals(
+ new Class<?>[] {
+ Integer.class, Integer.class, Integer.class,
Integer.class, Integer.class
+ },
+ TestUtils.getColumnDataTypes(inputTable));
+
+ Bucketizer bucketizer =
+ new Bucketizer()
+ .setInputCols("f1", "f2", "f3", "f4")
+ .setOutputCols("o1", "o2", "o3", "o4")
+ .setSplitsArray(splitsArray);
+
+ bucketizer.setHandleInvalid(HasHandleInvalid.SKIP_INVALID);
+ Table output = bucketizer.transform(inputTable)[0];
+
+ List<Row> expectedResult =
+ Arrays.asList(Row.of(1, 1.0, 1.0, 0.0, 0.0), Row.of(3, 1.0,
1.0, 0.0, 1.0));
+ verifyOutputResult(output, bucketizer.getOutputCols(), expectedResult);
+ }
+
@Test
public void testSaveLoadAndTransform() throws Exception {
Bucketizer bucketizer =
@@ -184,8 +211,7 @@ public class BucketizerTest extends AbstractTestBase {
.setHandleInvalid(HasHandleInvalid.KEEP_INVALID)
.setSplitsArray(splitsArray);
Bucketizer loadedBucketizer =
- StageTestUtils.saveAndReload(
- tEnv, bucketizer,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, bucketizer,
tempFolder.newFolder().getAbsolutePath());
Table output = loadedBucketizer.transform(inputTable)[0];
verifyOutputResult(output, loadedBucketizer.getOutputCols(),
expectedKeepResult);
}
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/MinMaxScalerTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/MinMaxScalerTest.java
index b26f042..1a0cfc4 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/MinMaxScalerTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/MinMaxScalerTest.java
@@ -24,9 +24,10 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.ml.feature.minmaxscaler.MinMaxScaler;
import org.apache.flink.ml.feature.minmaxscaler.MinMaxScalerModel;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -46,6 +47,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
/** Tests {@link MinMaxScaler} and {@link MinMaxScalerModel}. */
@@ -170,15 +172,31 @@ public class MinMaxScalerTest {
verifyPredictionResult(output, minMaxScaler.getOutputCol(),
EXPECTED_DATA);
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
trainDataTable);
+ predictDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
predictDataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
TestUtils.getColumnDataTypes(trainDataTable));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
+ TestUtils.getColumnDataTypes(predictDataTable));
+
+ MinMaxScaler minMaxScaler = new MinMaxScaler();
+ MinMaxScalerModel minMaxScalerModel = minMaxScaler.fit(trainDataTable);
+ Table output = minMaxScalerModel.transform(predictDataTable)[0];
+ verifyPredictionResult(output, minMaxScaler.getOutputCol(),
EXPECTED_DATA);
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
MinMaxScaler minMaxScaler = new MinMaxScaler();
MinMaxScaler loadedMinMaxScaler =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
tEnv, minMaxScaler,
tempFolder.newFolder().getAbsolutePath());
MinMaxScalerModel model = loadedMinMaxScaler.fit(trainDataTable);
MinMaxScalerModel loadedModel =
- StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Arrays.asList("minVector", "maxVector"),
model.getModelData()[0].getResolvedSchema().getColumnNames());
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/OneHotEncoderTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/OneHotEncoderTest.java
index b7a0758..8021648 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/OneHotEncoderTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/OneHotEncoderTest.java
@@ -28,7 +28,7 @@ import
org.apache.flink.ml.feature.onehotencoder.OneHotEncoderModelData;
import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -155,6 +155,21 @@ public class OneHotEncoderTest {
assertArrayEquals(expectedOutput, actualOutput);
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainTable = TestUtils.convertDataTypesToSparseInt(tEnv, trainTable);
+ predictTable = TestUtils.convertDataTypesToSparseInt(tEnv,
predictTable);
+ assertArrayEquals(new Class<?>[] {Integer.class},
TestUtils.getColumnDataTypes(trainTable));
+ assertArrayEquals(
+ new Class<?>[] {Integer.class},
TestUtils.getColumnDataTypes(predictTable));
+
+ OneHotEncoderModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ Map<Double, Vector>[] actualOutput =
+ executeAndCollect(outputTable, model.getInputCols(),
model.getOutputCols());
+ assertArrayEquals(expectedOutput, actualOutput);
+ }
+
@Test
public void testDropLast() {
estimator.setDropLast(false);
@@ -255,10 +270,9 @@ public class OneHotEncoderTest {
@Test
public void testSaveLoad() throws Exception {
estimator =
- StageTestUtils.saveAndReload(
- tEnv, estimator,
tempFolder.newFolder().getAbsolutePath());
+ TestUtils.saveAndReload(tEnv, estimator,
tempFolder.newFolder().getAbsolutePath());
OneHotEncoderModel model = estimator.fit(trainTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
Table outputTable = model.transform(predictTable)[0];
Map<Double, Vector>[] actualOutput =
executeAndCollect(outputTable, model.getInputCols(),
model.getOutputCols());
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StandardScalerTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StandardScalerTest.java
index 5337973..48a2664 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StandardScalerTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StandardScalerTest.java
@@ -24,10 +24,11 @@ import
org.apache.flink.ml.feature.standardscaler.StandardScaler;
import org.apache.flink.ml.feature.standardscaler.StandardScalerModel;
import org.apache.flink.ml.feature.standardscaler.StandardScalerModelData;
import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -182,15 +183,26 @@ public class StandardScalerTest extends AbstractTestBase {
verifyPredictionResult(expectedResWithMeanAndStd, output,
standardScaler.getOutputCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ denseTable = TestUtils.convertDataTypesToSparseInt(tEnv, denseTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
TestUtils.getColumnDataTypes(denseTable));
+
+ StandardScaler standardScaler = new StandardScaler().setWithMean(true);
+ Table output = standardScaler.fit(denseTable).transform(denseTable)[0];
+ verifyPredictionResult(expectedResWithMeanAndStd, output,
standardScaler.getOutputCol());
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
StandardScaler standardScaler = new StandardScaler();
standardScaler =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
tEnv, standardScaler,
tempFolder.newFolder().getAbsolutePath());
StandardScalerModel model = standardScaler.fit(denseTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Arrays.asList("mean", "std"),
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VectorAssemblerTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VectorAssemblerTest.java
index 15fd5d9..420844b 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VectorAssemblerTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VectorAssemblerTest.java
@@ -25,7 +25,7 @@ import
org.apache.flink.ml.feature.vectorassembler.VectorAssembler;
import org.apache.flink.ml.linalg.DenseVector;
import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -170,7 +170,28 @@ public class VectorAssemblerTest extends AbstractTestBase {
.setOutputCol("assembledVec")
.setHandleInvalid(HasHandleInvalid.SKIP_INVALID);
VectorAssembler loadedVectorAssembler =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
+ tEnv, vectorAssembler,
TEMPORARY_FOLDER.newFolder().getAbsolutePath());
+ Table output = loadedVectorAssembler.transform(inputDataTable)[0];
+ verifyOutputResult(output, loadedVectorAssembler.getOutputCol(), 2);
+ }
+
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ inputDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
inputDataTable);
+ assertArrayEquals(
+ new Class<?>[] {
+ Integer.class, SparseVector.class, Integer.class,
SparseVector.class
+ },
+ TestUtils.getColumnDataTypes(inputDataTable));
+
+ VectorAssembler vectorAssembler =
+ new VectorAssembler()
+ .setInputCols("vec", "num", "sparseVec")
+ .setOutputCol("assembledVec")
+ .setHandleInvalid(HasHandleInvalid.SKIP_INVALID);
+ VectorAssembler loadedVectorAssembler =
+ TestUtils.saveAndReload(
tEnv, vectorAssembler,
TEMPORARY_FOLDER.newFolder().getAbsolutePath());
Table output = loadedVectorAssembler.transform(inputDataTable)[0];
verifyOutputResult(output, loadedVectorAssembler.getOutputCol(), 2);
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/IndexToStringModelTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/IndexToStringModelTest.java
index bc482a1..937b829 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/IndexToStringModelTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/IndexToStringModelTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.ml.feature.stringindexer;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -137,7 +137,7 @@ public class IndexToStringModelTest extends
AbstractTestBase {
.setInputCols("inputCol1", "inputCol2")
.setOutputCols("outputCol1", "outputCol2")
.setModelData(modelTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Collections.singletonList("stringArrays"),
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/StringIndexerTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/StringIndexerTest.java
index c8af00e..08b36a1 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/StringIndexerTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/stringindexer/StringIndexerTest.java
@@ -22,7 +22,7 @@ import
org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.ml.common.param.HasHandleInvalid;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -59,13 +59,25 @@ public class StringIndexerTest extends AbstractTestBase {
private final String[][] expectedAlphabeticAscModelData =
new String[][] {{"a", "b", "c", "d"}, {"-1.0", "0.0", "1.0",
"2.0"}};
private final List<Row> expectedAlphabeticAscPredictData =
- Arrays.asList(Row.of("a", 2.0, 0, 3), Row.of("b", 1.0, 1, 2),
Row.of("e", 2.0, 4, 3));
+ Arrays.asList(
+ Row.of("a", 2.0, 0.0, 3.0),
+ Row.of("b", 1.0, 1.0, 2.0),
+ Row.of("e", 2.0, 4.0, 3.0));
private final List<Row> expectedAlphabeticDescPredictData =
- Arrays.asList(Row.of("a", 2.0, 3, 0), Row.of("b", 1.0, 2, 1),
Row.of("e", 2.0, 4, 0));
+ Arrays.asList(
+ Row.of("a", 2.0, 3.0, 0.0),
+ Row.of("b", 1.0, 2.0, 1.0),
+ Row.of("e", 2.0, 4.0, 0.0));
private final List<Row> expectedFreqAscPredictData =
- Arrays.asList(Row.of("a", 2.0, 2, 3), Row.of("b", 1.0, 3, 1),
Row.of("e", 2.0, 4, 3));
+ Arrays.asList(
+ Row.of("a", 2.0, 2.0, 3.0),
+ Row.of("b", 1.0, 3.0, 1.0),
+ Row.of("e", 2.0, 4.0, 3.0));
private final List<Row> expectedFreqDescPredictData =
- Arrays.asList(Row.of("a", 2.0, 1, 0), Row.of("b", 1.0, 0, 2),
Row.of("e", 2.0, 4, 0));
+ Arrays.asList(
+ Row.of("a", 2.0, 1.0, 0.0),
+ Row.of("b", 1.0, 0.0, 2.0),
+ Row.of("e", 2.0, 4.0, 0.0));
@Before
public void before() {
@@ -170,14 +182,14 @@ public class StringIndexerTest extends AbstractTestBase {
output = stringIndexer.fit(trainTable).transform(predictTable)[0];
predictedResult =
IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
- Set<Integer> distinctStringsCol1 = new HashSet<>();
- Set<Integer> distinctStringsCol2 = new HashSet<>();
- int index;
+ Set<Double> distinctStringsCol1 = new HashSet<>();
+ Set<Double> distinctStringsCol2 = new HashSet<>();
+ double index;
for (Row r : predictedResult) {
- index = (Integer) r.getField(2);
+ index = (Double) r.getField(2);
distinctStringsCol1.add(index);
assertTrue(index >= 0 && index <= 4);
- index = (Integer) r.getField(3);
+ index = (Double) r.getField(3);
assertTrue(index >= 0 && index <= 3);
distinctStringsCol2.add(index);
}
@@ -209,7 +221,7 @@ public class StringIndexerTest extends AbstractTestBase {
stringIndexer.setHandleInvalid(StringIndexerParams.SKIP_INVALID);
output = stringIndexer.fit(trainTable).transform(predictTable)[0];
predictedResult =
IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
- expectedResult = Arrays.asList(Row.of("a", 2.0, 0, 3), Row.of("b",
1.0, 1, 2));
+ expectedResult = Arrays.asList(Row.of("a", 2.0, 0.0, 3.0), Row.of("b",
1.0, 1.0, 2.0));
verifyPredictionResult(expectedResult, predictedResult);
// Throws an exception on invalid data.
@@ -254,11 +266,11 @@ public class StringIndexerTest extends AbstractTestBase {
.setStringOrderType(StringIndexerParams.ALPHABET_ASC_ORDER)
.setHandleInvalid(StringIndexerParams.KEEP_INVALID);
stringIndexer =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
tEnv, stringIndexer,
tempFolder.newFolder().getAbsolutePath());
StringIndexerModel model = stringIndexer.fit(trainTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Collections.singletonList("stringArrays"),
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/regression/LinearRegressionTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/regression/LinearRegressionTest.java
index 3ea99d3..f10a63b 100644
---
a/flink-ml-lib/src/test/java/org/apache/flink/ml/regression/LinearRegressionTest.java
+++
b/flink-ml-lib/src/test/java/org/apache/flink/ml/regression/LinearRegressionTest.java
@@ -23,13 +23,14 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.linalg.SparseVector;
import org.apache.flink.ml.linalg.Vectors;
import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
import org.apache.flink.ml.regression.linearregression.LinearRegression;
import org.apache.flink.ml.regression.linearregression.LinearRegressionModel;
import
org.apache.flink.ml.regression.linearregression.LinearRegressionModelData;
import org.apache.flink.ml.util.ReadWriteUtils;
-import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.ml.util.TestUtils;
import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
@@ -109,7 +110,7 @@ public class LinearRegressionTest {
throws Exception {
List<Row> predResult =
IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
for (Row predictionRow : predResult) {
- double label = (double) predictionRow.getField(labelCol);
+ double label = ((Number)
predictionRow.getField(labelCol)).doubleValue();
double prediction = (double) predictionRow.getField(predictionCol);
assertTrue(Math.abs(prediction - label) / label <
PREDICTION_TOLERANCE);
}
@@ -175,14 +176,27 @@ public class LinearRegressionTest {
output, linearRegression.getLabelCol(),
linearRegression.getPredictionCol());
}
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainDataTable = TestUtils.convertDataTypesToSparseInt(tEnv,
trainDataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class, Integer.class,
Integer.class},
+ TestUtils.getColumnDataTypes(trainDataTable));
+
+ LinearRegression linearRegression = new
LinearRegression().setWeightCol("weight");
+ Table output =
linearRegression.fit(trainDataTable).transform(trainDataTable)[0];
+ verifyPredictionResult(
+ output, linearRegression.getLabelCol(),
linearRegression.getPredictionCol());
+ }
+
@Test
public void testSaveLoadAndPredict() throws Exception {
LinearRegression linearRegression = new
LinearRegression().setWeightCol("weight");
linearRegression =
- StageTestUtils.saveAndReload(
+ TestUtils.saveAndReload(
tEnv, linearRegression,
tempFolder.newFolder().getAbsolutePath());
LinearRegressionModel model = linearRegression.fit(trainDataTable);
- model = StageTestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
+ model = TestUtils.saveAndReload(tEnv, model,
tempFolder.newFolder().getAbsolutePath());
assertEquals(
Collections.singletonList("coefficient"),
model.getModelData()[0].getResolvedSchema().getColumnNames());
diff --git
a/flink-ml-lib/src/test/java/org/apache/flink/ml/util/StageTestUtils.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/util/StageTestUtils.java
deleted file mode 100644
index 138a6fc..0000000
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/util/StageTestUtils.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.util;
-
-import org.apache.flink.ml.api.Stage;
-import org.apache.flink.ml.common.datastream.TableUtils;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
-
-import java.lang.reflect.Method;
-
-/** Utility methods for testing stages. */
-public class StageTestUtils {
- /**
- * Saves a stage to filesystem and reloads it by invoking the static
load() method of the given
- * stage.
- */
- public static <T extends Stage<T>> T saveAndReload(
- StreamTableEnvironment tEnv, T stage, String path) throws
Exception {
- StreamExecutionEnvironment env =
TableUtils.getExecutionEnvironment(tEnv);
-
- stage.save(path);
- try {
- env.execute();
- } catch (RuntimeException e) {
- if (!e.getMessage()
- .equals("No operators defined in streaming topology.
Cannot execute.")) {
- throw e;
- }
- }
-
- Method method =
- stage.getClass().getMethod("load",
StreamTableEnvironment.class, String.class);
- return (T) method.invoke(null, tEnv, path);
- }
-}
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/util/TestUtils.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/util/TestUtils.java
new file mode 100644
index 0000000..171aa66
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/util/TestUtils.java
@@ -0,0 +1,117 @@
+/*
+ * 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.util;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.Stage;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+import org.apache.flink.ml.linalg.typeinfo.SparseVectorTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.lang.reflect.Method;
+
+/** Utility methods for unit tests. */
+public class TestUtils {
+ /**
+ * Saves a stage to filesystem and reloads it by invoking the static
load() method of the given
+ * stage.
+ */
+ public static <T extends Stage<T>> T saveAndReload(
+ StreamTableEnvironment tEnv, T stage, String path) throws
Exception {
+ StreamExecutionEnvironment env =
TableUtils.getExecutionEnvironment(tEnv);
+
+ stage.save(path);
+ try {
+ env.execute();
+ } catch (RuntimeException e) {
+ if (!e.getMessage()
+ .equals("No operators defined in streaming topology.
Cannot execute.")) {
+ throw e;
+ }
+ }
+
+ Method method =
+ stage.getClass().getMethod("load",
StreamTableEnvironment.class, String.class);
+ return (T) method.invoke(null, tEnv, path);
+ }
+
+ /**
+ * Converts data types in the table to sparse types and integer types.
+ *
+ * <ul>
+ * <li>If a column in the table is of DenseVector type, converts it to
SparseVector.
+ * <li>If a column in the table is of Double type, converts it to
integer.
+ * </ul>
+ */
+ public static Table convertDataTypesToSparseInt(StreamTableEnvironment
tEnv, Table table) {
+ RowTypeInfo inputTypeInfo =
TableUtils.getRowTypeInfo(table.getResolvedSchema());
+ TypeInformation<?>[] fieldTypes = inputTypeInfo.getFieldTypes();
+ for (int i = 0; i < fieldTypes.length; i++) {
+ if (fieldTypes[i].equals(DenseVectorTypeInfo.INSTANCE)) {
+ fieldTypes[i] = SparseVectorTypeInfo.INSTANCE;
+ } else if (fieldTypes[i].equals(Types.DOUBLE)) {
+ fieldTypes[i] = Types.INT;
+ }
+ }
+
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(fieldTypes),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames()));
+ DataStream<Row> dataStream = tEnv.toDataStream(table);
+ dataStream =
+ dataStream.map(
+ new MapFunction<Row, Row>() {
+ @Override
+ public Row map(Row row) {
+ int arity = row.getArity();
+ for (int i = 0; i < arity; i++) {
+ Object obj = row.getField(i);
+ if (obj instanceof Vector) {
+ row.setField(i, ((Vector)
obj).toSparse());
+ } else if (obj instanceof Number) {
+ row.setField(i, ((Number)
obj).intValue());
+ }
+ }
+ return row;
+ }
+ },
+ outputTypeInfo);
+ return tEnv.fromDataStream(dataStream);
+ }
+
+ /** Gets the types of data in each column of the input table. */
+ public static Class<?>[] getColumnDataTypes(Table table) {
+ return table.getResolvedSchema().getColumnDataTypes().stream()
+ .map(DataType::getConversionClass)
+ .toArray(Class<?>[]::new);
+ }
+}
