This is an automated email from the ASF dual-hosted git repository.
zhangzp 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 f6b071f [hotfix] Fix finding buckets error in Bucketizer (#96)
f6b071f is described below
commit f6b071fb6832eaac9d180e7a9ab0ee57e4dcbe2d
Author: Mr-Mu <[email protected]>
AuthorDate: Tue May 10 03:14:07 2022 -0500
[hotfix] Fix finding buckets error in Bucketizer (#96)
This closes #96.
---
.../flink/ml/feature/bucketizer/Bucketizer.java | 4 +--
.../apache/flink/ml/feature/BucketizerTest.java | 38 ++++++++++++----------
2 files changed, 22 insertions(+), 20 deletions(-)
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 836e30b..8afefcb 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
@@ -110,13 +110,13 @@ public class Bucketizer implements
Transformer<Bucketizer>, BucketizerParams<Buc
if (!Double.isNaN(feature)) {
int index = Arrays.binarySearch(splits, feature);
if (index >= 0) {
- if (index == inputCols.length - 1) {
+ if (index == splits.length - 1) {
index--;
}
outputRow.setField(i, index);
} else {
index = -index - 1;
- if (index == 0 || index == inputCols.length) {
+ if (index == 0 || index == splits.length) {
isInvalid = true;
} else {
outputRow.setField(i, index - 1);
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 4a7306c..7170103 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
@@ -55,21 +55,22 @@ public class BucketizerTest extends AbstractTestBase {
private static final List<Row> inputData =
Arrays.asList(
- Row.of(1, -0.5, 0.0, 1.0),
- Row.of(2, Double.NEGATIVE_INFINITY, 1.0,
Double.POSITIVE_INFINITY),
- Row.of(3, Double.NaN, -0.5, -0.5));
+ Row.of(1, -0.5, 0.0, 1.0, 0.0),
+ Row.of(2, Double.NEGATIVE_INFINITY, 1.0,
Double.POSITIVE_INFINITY, 1.0),
+ Row.of(3, Double.NaN, -0.5, -0.5, 2.0));
private static final Double[][] splitsArray =
new Double[][] {
new Double[] {-0.5, 0.0, 0.5},
new Double[] {-1.0, 0.0, 2.0},
- new Double[] {Double.NEGATIVE_INFINITY, 10.0,
Double.POSITIVE_INFINITY}
+ new Double[] {Double.NEGATIVE_INFINITY, 10.0,
Double.POSITIVE_INFINITY},
+ new Double[] {Double.NEGATIVE_INFINITY, 1.5,
Double.POSITIVE_INFINITY}
};
private final List<Row> expectedKeepResult =
- Arrays.asList(Row.of(1, 0, 1, 0), Row.of(2, 2, 1, 1), Row.of(3, 2,
0, 0));
+ Arrays.asList(Row.of(1, 0, 1, 0, 0), Row.of(2, 2, 1, 1, 0),
Row.of(3, 2, 0, 0, 1));
- private final List<Row> expectedSkipResult =
Collections.singletonList(Row.of(1, 0, 1, 0));
+ private final List<Row> expectedSkipResult =
Collections.singletonList(Row.of(1, 0, 1, 0, 0));
@Before
public void before() {
@@ -80,7 +81,8 @@ public class BucketizerTest extends AbstractTestBase {
env.enableCheckpointing(100);
env.setRestartStrategy(RestartStrategies.noRestart());
tEnv = StreamTableEnvironment.create(env);
- inputTable =
tEnv.fromDataStream(env.fromCollection(inputData)).as("id", "f1", "f2", "f3");
+ inputTable =
+ tEnv.fromDataStream(env.fromCollection(inputData)).as("id",
"f1", "f2", "f3", "f4");
}
@SuppressWarnings("all")
@@ -109,12 +111,12 @@ public class BucketizerTest extends AbstractTestBase {
assertEquals(HasHandleInvalid.ERROR_INVALID,
bucketizer.getHandleInvalid());
bucketizer
- .setInputCols("f1", "f2", "f3")
- .setOutputCols("o1", "o2", "o3")
+ .setInputCols("f1", "f2", "f3", "f4")
+ .setOutputCols("o1", "o2", "o3", "o4")
.setHandleInvalid(HasHandleInvalid.SKIP_INVALID)
.setSplitsArray(splitsArray);
- assertArrayEquals(new String[] {"f1", "f2", "f3"},
bucketizer.getInputCols());
- assertArrayEquals(new String[] {"o1", "o2", "o3"},
bucketizer.getOutputCols());
+ assertArrayEquals(new String[] {"f1", "f2", "f3", "f4"},
bucketizer.getInputCols());
+ assertArrayEquals(new String[] {"o1", "o2", "o3", "o4"},
bucketizer.getOutputCols());
assertEquals(HasHandleInvalid.SKIP_INVALID,
bucketizer.getHandleInvalid());
Double[][] setSplitsArray = bucketizer.getSplitsArray();
@@ -128,13 +130,13 @@ public class BucketizerTest extends AbstractTestBase {
public void testOutputSchema() {
Bucketizer bucketizer =
new Bucketizer()
- .setInputCols("f1", "f2", "f3")
- .setOutputCols("o1", "o2", "o3")
+ .setInputCols("f1", "f2", "f3", "f4")
+ .setOutputCols("o1", "o2", "o3", "o4")
.setHandleInvalid(HasHandleInvalid.SKIP_INVALID)
.setSplitsArray(splitsArray);
Table output = bucketizer.transform(inputTable)[0];
assertEquals(
- Arrays.asList("id", "f1", "f2", "f3", "o1", "o2", "o3"),
+ Arrays.asList("id", "f1", "f2", "f3", "f4", "o1", "o2", "o3",
"o4"),
output.getResolvedSchema().getColumnNames());
}
@@ -142,8 +144,8 @@ public class BucketizerTest extends AbstractTestBase {
public void testTransform() throws Exception {
Bucketizer bucketizer =
new Bucketizer()
- .setInputCols("f1", "f2", "f3")
- .setOutputCols("o1", "o2", "o3")
+ .setInputCols("f1", "f2", "f3", "f4")
+ .setOutputCols("o1", "o2", "o3", "o4")
.setSplitsArray(splitsArray);
Table output;
@@ -177,8 +179,8 @@ public class BucketizerTest extends AbstractTestBase {
public void testSaveLoadAndTransform() throws Exception {
Bucketizer bucketizer =
new Bucketizer()
- .setInputCols("f1", "f2", "f3")
- .setOutputCols("o1", "o2", "o3")
+ .setInputCols("f1", "f2", "f3", "f4")
+ .setOutputCols("o1", "o2", "o3", "o4")
.setHandleInvalid(HasHandleInvalid.KEEP_INVALID)
.setSplitsArray(splitsArray);
Bucketizer loadedBucketizer =
