You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2021/10/27 15:41:07 UTC
[flink-ml] 01/02: [hotfix] upgrade to FLINK1.14 and delete outdated
codes about dataset
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
commit fd820ea275baed2068bc2e17e92fa3f3fb6ad6cb
Author: zhangzp <zh...@gmail.com>
AuthorDate: Tue Sep 28 18:01:03 2021 +0800
[hotfix] upgrade to FLINK1.14 and delete outdated codes about dataset
---
.../org/apache/flink/ml/common/MLEnvironment.java | 66 +------
.../basicstatistic/MultivariateGaussian.java | 7 +-
.../ml/common/utils/DataSetConversionUtil.java | 167 -----------------
.../apache/flink/ml/common/utils/TableUtil.java | 15 --
.../apache/flink/ml/common/MLEnvironmentTest.java | 20 +--
.../ml/common/utils/DataSetConversionUtilTest.java | 200 ---------------------
pom.xml | 2 +-
7 files changed, 8 insertions(+), 469 deletions(-)
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/MLEnvironment.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/MLEnvironment.java
index cff7ecd..aea7a67 100644
--- a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/MLEnvironment.java
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/MLEnvironment.java
@@ -22,7 +22,6 @@ package org.apache.flink.ml.common;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/**
@@ -34,30 +33,15 @@ import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
*
* @see ExecutionEnvironment
* @see StreamExecutionEnvironment
- * @see BatchTableEnvironment
* @see StreamTableEnvironment
*/
public class MLEnvironment {
- private ExecutionEnvironment env;
private StreamExecutionEnvironment streamEnv;
- private BatchTableEnvironment batchTableEnv;
private StreamTableEnvironment streamTableEnv;
/** Construct with null that the class can load the environment in the `get` method. */
public MLEnvironment() {
- this(null, null, null, null);
- }
-
- /**
- * Construct with the batch environment and the the batch table environment given by user.
- *
- * <p>The env can be null which will be loaded in the `get` method.
- *
- * @param batchEnv the ExecutionEnvironment
- * @param batchTableEnv the BatchTableEnvironment
- */
- public MLEnvironment(ExecutionEnvironment batchEnv, BatchTableEnvironment batchTableEnv) {
- this(batchEnv, batchTableEnv, null, null);
+ this(null, null);
}
/**
@@ -70,44 +54,11 @@ public class MLEnvironment {
*/
public MLEnvironment(
StreamExecutionEnvironment streamEnv, StreamTableEnvironment streamTableEnv) {
- this(null, null, streamEnv, streamTableEnv);
- }
-
- /**
- * Construct with env given by user.
- *
- * <p>The env can be null which will be loaded in the `get` method.
- *
- * @param batchEnv the ExecutionEnvironment
- * @param batchTableEnv the BatchTableEnvironment
- * @param streamEnv the StreamExecutionEnvironment
- * @param streamTableEnv the StreamTableEnvironment
- */
- public MLEnvironment(
- ExecutionEnvironment batchEnv,
- BatchTableEnvironment batchTableEnv,
- StreamExecutionEnvironment streamEnv,
- StreamTableEnvironment streamTableEnv) {
- this.env = batchEnv;
- this.batchTableEnv = batchTableEnv;
this.streamEnv = streamEnv;
this.streamTableEnv = streamTableEnv;
}
/**
- * Get the ExecutionEnvironment. if the ExecutionEnvironment has not been set, it initial the
- * ExecutionEnvironment with default Configuration.
- *
- * @return the batch {@link ExecutionEnvironment}
- */
- public ExecutionEnvironment getExecutionEnvironment() {
- if (null == env) {
- env = ExecutionEnvironment.getExecutionEnvironment();
- }
- return env;
- }
-
- /**
* Get the StreamExecutionEnvironment. if the StreamExecutionEnvironment has not been set, it
* initial the StreamExecutionEnvironment with default Configuration.
*
@@ -121,19 +72,6 @@ public class MLEnvironment {
}
/**
- * Get the BatchTableEnvironment. if the BatchTableEnvironment has not been set, it initial the
- * BatchTableEnvironment with default Configuration.
- *
- * @return the {@link BatchTableEnvironment}
- */
- public BatchTableEnvironment getBatchTableEnvironment() {
- if (null == batchTableEnv) {
- batchTableEnv = BatchTableEnvironment.create(getExecutionEnvironment());
- }
- return batchTableEnv;
- }
-
- /**
* Get the StreamTableEnvironment. if the StreamTableEnvironment has not been set, it initial
* the StreamTableEnvironment with default Configuration.
*
@@ -144,7 +82,7 @@ public class MLEnvironment {
streamTableEnv =
StreamTableEnvironment.create(
getStreamExecutionEnvironment(),
- EnvironmentSettings.newInstance().useOldPlanner().build());
+ EnvironmentSettings.newInstance().build());
}
return streamTableEnv;
}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/statistics/basicstatistic/MultivariateGaussian.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/statistics/basicstatistic/MultivariateGaussian.java
index 1f5f806..bf3a7b9 100644
--- a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/statistics/basicstatistic/MultivariateGaussian.java
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/statistics/basicstatistic/MultivariateGaussian.java
@@ -25,8 +25,6 @@ import org.apache.flink.ml.common.linalg.DenseVector;
import org.apache.flink.ml.common.linalg.SparseVector;
import org.apache.flink.ml.common.linalg.Vector;
-import org.apache.flink.shaded.guava18.com.google.common.primitives.Doubles;
-
import com.github.fommil.netlib.LAPACK;
import org.netlib.util.intW;
@@ -116,7 +114,10 @@ public class MultivariateGaussian {
System.arraycopy(cov.getData(), 0, matU, 0, k * k);
LAPACK_INST.dsyev("V", "U", k, matU, k, evs, work, lwork, info);
- double maxEv = Doubles.max(evs);
+ double maxEv = Double.MIN_VALUE;
+ for (double ev : evs) {
+ maxEv = Math.max(maxEv, ev);
+ }
double tol = EPSILON * k * maxEv;
// log(pseudo-determinant) is sum of the logs of all non-zero singular values
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/DataSetConversionUtil.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/DataSetConversionUtil.java
deleted file mode 100644
index f146a9f..0000000
--- a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/DataSetConversionUtil.java
+++ /dev/null
@@ -1,167 +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.common.utils;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.operators.SingleInputUdfOperator;
-import org.apache.flink.api.java.operators.TwoInputUdfOperator;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.ml.common.MLEnvironment;
-import org.apache.flink.ml.common.MLEnvironmentFactory;
-import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.types.Row;
-
-/** Provide functions of conversions between DataSet and Table. */
-public class DataSetConversionUtil {
- /**
- * Convert the given Table to {@link DataSet}<{@link Row}>.
- *
- * @param sessionId the sessionId of {@link MLEnvironmentFactory}
- * @param table the Table to convert.
- * @return the converted DataSet.
- */
- public static DataSet<Row> fromTable(Long sessionId, Table table) {
- return MLEnvironmentFactory.get(sessionId)
- .getBatchTableEnvironment()
- .toDataSet(table, Row.class);
- }
-
- /**
- * Convert the given DataSet into a Table with specified TableSchema.
- *
- * @param sessionId the sessionId of {@link MLEnvironmentFactory}
- * @param data the DataSet to convert.
- * @param schema the specified TableSchema.
- * @return the converted Table.
- */
- public static Table toTable(Long sessionId, DataSet<Row> data, TableSchema schema) {
- // TableSchema.getFieldTypes() is deprecated, this should be improved once FLIP-65 is fully
- // merged.
- return toTable(sessionId, data, schema.getFieldNames(), schema.getFieldTypes());
- }
-
- /**
- * Convert the given DataSet into a Table with specified colNames and colTypes.
- *
- * @param sessionId sessionId the sessionId of {@link MLEnvironmentFactory}.
- * @param data the DataSet to convert.
- * @param colNames the specified colNames.
- * @param colTypes the specified colTypes. This variable is used only when the DataSet is
- * produced by a function and Flink cannot determine automatically what the produced type
- * is.
- * @return the converted Table.
- */
- public static Table toTable(
- Long sessionId, DataSet<Row> data, String[] colNames, TypeInformation<?>[] colTypes) {
- return toTable(MLEnvironmentFactory.get(sessionId), data, colNames, colTypes);
- }
-
- /**
- * Convert the given DataSet into a Table with specified colNames.
- *
- * @param sessionId sessionId the sessionId of {@link MLEnvironmentFactory}.
- * @param data the DataSet to convert.
- * @param colNames the specified colNames.
- * @return the converted Table.
- */
- public static Table toTable(Long sessionId, DataSet<Row> data, String[] colNames) {
- return toTable(MLEnvironmentFactory.get(sessionId), data, colNames);
- }
-
- /**
- * Convert the given DataSet into a Table with specified colNames and colTypes.
- *
- * @param session the MLEnvironment using to convert DataSet to Table.
- * @param data the DataSet to convert.
- * @param colNames the specified colNames.
- * @param colTypes the specified colTypes. This variable is used only when the DataSet is
- * produced by a function and Flink cannot determine automatically what the produced type
- * is.
- * @return the converted Table.
- */
- public static Table toTable(
- MLEnvironment session,
- DataSet<Row> data,
- String[] colNames,
- TypeInformation<?>[] colTypes) {
- try {
- if (null != colTypes) {
- // Try to add row type information for the dataset to be converted.
- // In most case, this keeps us from the rolling back logic in the catch block,
- // which adds an unnecessary map function just in order to add row type information.
- if (data instanceof SingleInputUdfOperator) {
- ((SingleInputUdfOperator) data).returns(new RowTypeInfo(colTypes, colNames));
- } else if (data instanceof TwoInputUdfOperator) {
- ((TwoInputUdfOperator) data).returns(new RowTypeInfo(colTypes, colNames));
- }
- }
- return toTable(session, data, colNames);
- } catch (ValidationException ex) {
- // currently ValidationException will be thrown and caught for further processing.
- // Because the getType() API of the Transformation can only be accessed once.
- // This can be improve if we add "isTypeSet()" API to the Transformation class.
- if (null == colTypes) {
- throw ex;
- } else {
- DataSet<Row> t = fallbackToExplicitTypeDefine(data, colNames, colTypes);
- return toTable(session, t, colNames);
- }
- }
- }
-
- /**
- * Convert the given DataSet into a Table with specified colNames.
- *
- * @param session the MLEnvironment using to convert DataSet to Table.
- * @param data the DataSet to convert.
- * @param colNames the specified colNames.
- * @return the converted Table.
- */
- public static Table toTable(MLEnvironment session, DataSet<Row> data, String[] colNames) {
- if (null == colNames || colNames.length == 0) {
- return session.getBatchTableEnvironment().fromDataSet(data);
- } else {
- StringBuilder sbd = new StringBuilder();
- sbd.append(colNames[0]);
- for (int i = 1; i < colNames.length; i++) {
- sbd.append(",").append(colNames[i]);
- }
- return session.getBatchTableEnvironment().fromDataSet(data, sbd.toString());
- }
- }
-
- /**
- * Adds a type information hint about the colTypes with the Row to the DataSet.
- *
- * @param data the DataSet to add type information.
- * @param colNames the specified colNames
- * @param colTypes the specified colTypes
- * @return the DataSet with type information hint.
- */
- private static DataSet<Row> fallbackToExplicitTypeDefine(
- DataSet<Row> data, String[] colNames, TypeInformation<?>[] colTypes) {
- return data.map((MapFunction<Row, Row>) t -> t)
- .returns(new RowTypeInfo(colTypes, colNames));
- }
-}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/TableUtil.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/TableUtil.java
index 0cfd396..3f2261e 100644
--- a/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/TableUtil.java
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/common/utils/TableUtil.java
@@ -25,8 +25,6 @@ import org.apache.flink.table.api.TableSchema;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.shaded.guava18.com.google.common.base.Joiner;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -423,17 +421,4 @@ public class TableUtil {
return sbd.toString();
}
-
- /**
- * Convert column name array to SQL clause.
- *
- * <p>For example, columns "{a, b, c}" will be converted into a SQL-compatible select string
- * section: "`a`, `b`, `c`".
- *
- * @param colNames columns to convert
- * @return converted SQL clause.
- */
- public static String columnsToSqlClause(String[] colNames) {
- return Joiner.on("`,`").appendTo(new StringBuilder("`"), colNames).append("`").toString();
- }
}
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/common/MLEnvironmentTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/common/MLEnvironmentTest.java
index bbc1c34..db51622 100644
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/common/MLEnvironmentTest.java
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/common/MLEnvironmentTest.java
@@ -19,10 +19,8 @@
package org.apache.flink.ml.common;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Assert;
@@ -35,33 +33,17 @@ public class MLEnvironmentTest {
@Test
public void testDefaultConstructor() {
MLEnvironment mlEnvironment = new MLEnvironment();
- Assert.assertNotNull(mlEnvironment.getExecutionEnvironment());
- Assert.assertNotNull(mlEnvironment.getBatchTableEnvironment());
Assert.assertNotNull(mlEnvironment.getStreamExecutionEnvironment());
Assert.assertNotNull(mlEnvironment.getStreamTableEnvironment());
}
@Test
- public void testConstructWithBatchEnv() {
- ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
- BatchTableEnvironment batchTableEnvironment =
- BatchTableEnvironment.create(executionEnvironment);
-
- MLEnvironment mlEnvironment =
- new MLEnvironment(executionEnvironment, batchTableEnvironment);
-
- Assert.assertSame(mlEnvironment.getExecutionEnvironment(), executionEnvironment);
- Assert.assertSame(mlEnvironment.getBatchTableEnvironment(), batchTableEnvironment);
- }
-
- @Test
public void testConstructWithStreamEnv() {
StreamExecutionEnvironment streamExecutionEnvironment =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment streamTableEnvironment =
StreamTableEnvironment.create(
- streamExecutionEnvironment,
- EnvironmentSettings.newInstance().useOldPlanner().build());
+ streamExecutionEnvironment, EnvironmentSettings.newInstance().build());
MLEnvironment mlEnvironment =
new MLEnvironment(streamExecutionEnvironment, streamTableEnvironment);
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/common/utils/DataSetConversionUtilTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/common/utils/DataSetConversionUtilTest.java
deleted file mode 100644
index 4b84569..0000000
--- a/flink-ml-lib/src/test/java/org/apache/flink/ml/common/utils/DataSetConversionUtilTest.java
+++ /dev/null
@@ -1,200 +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.common.utils;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.ml.common.MLEnvironmentFactory;
-import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.types.Row;
-
-import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.ExecutionException;
-
-/** Unit test for DataSetConversionUtil. */
-public class DataSetConversionUtilTest {
- @Rule public ExpectedException thrown = ExpectedException.none();
-
- @Test
- public void testForceTypeSchema() {
- ExecutionEnvironment env = MLEnvironmentFactory.getDefault().getExecutionEnvironment();
-
- DataSet<Row> input = env.fromElements(Row.of("s1")).map(new GenericTypeMap());
- Table table2 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- input,
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)});
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}),
- table2.getSchema());
- }
-
- @Test
- public void testForceTypeWithTableSchema() {
- ExecutionEnvironment env = MLEnvironmentFactory.getDefault().getExecutionEnvironment();
-
- DataSet<Row> input = env.fromElements(Row.of("s1")).map(new GenericTypeMap());
- Table table2 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- input,
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}));
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}),
- table2.getSchema());
- }
-
- @Test
- public void testExceptionWithoutTypeSchema() {
- thrown.expect(ValidationException.class);
- ExecutionEnvironment env = MLEnvironmentFactory.getDefault().getExecutionEnvironment();
- DataSet<Row> input = env.fromElements(Row.of("s1")).map(new GenericTypeMap());
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID, input, new String[] {"f0"});
- }
-
- @Test
- public void testBasicConvert() throws Exception {
- ExecutionEnvironment env = MLEnvironmentFactory.getDefault().getExecutionEnvironment();
-
- DataSet<Row> input = env.fromElements(Row.of("a"));
-
- Table table1 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- input,
- new String[] {"word"});
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(String.class)}),
- table1.getSchema());
- List<Row> list =
- DataSetConversionUtil.fromTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID, table1)
- .collect();
- Assert.assertEquals(Collections.singletonList(Row.of("a")), list);
- }
-
- @Test
- public void testE2E() throws Exception {
- ExecutionEnvironment env = MLEnvironmentFactory.getDefault().getExecutionEnvironment();
-
- DataSet<Row> input = env.fromElements(Row.of("a"));
-
- Table table1 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- input,
- new String[] {"word"});
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(String.class)}),
- table1.getSchema());
-
- DataSet<Row> genericInput1 = input.map(new GenericTypeMap());
-
- // Force type should go through with explicit type info.
- Table table2 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- genericInput1,
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)});
-
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}),
- table2.getSchema());
-
- DataSet<Row> genericInput2 = input.map(new GenericTypeMap());
-
- // Force type should go through with table schema.
- Table table3 =
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- genericInput2,
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}));
-
- Assert.assertEquals(
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}),
- table3.getSchema());
-
- // applying toTable again on the same input should fail
- thrown.expect(IllegalStateException.class);
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- genericInput2,
- new TableSchema(
- new String[] {"word"},
- new TypeInformation[] {TypeInformation.of(Integer.class)}));
-
- // Validation should fail without correct type inference.
- DataSet<Row> genericInput3 = input.map(new GenericTypeMap());
- thrown.expect(ValidationException.class);
- DataSetConversionUtil.toTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,
- genericInput3,
- new String[] {"word"});
-
- // Output should go through when using correct type to output.
- DataSet<Row> output =
- DataSetConversionUtil.fromTable(
- MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID, table1);
- output.print();
-
- // Output should NOT go through when using incorrect type forcing.
- thrown.expect(ExecutionException.class);
- DataSetConversionUtil.fromTable(MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID, table2)
- .print();
- }
-
- private static class GenericTypeMap implements MapFunction<Row, Row> {
-
- @Override
- public Row map(Row value) throws Exception {
- return value;
- }
- }
-}
diff --git a/pom.xml b/pom.xml
index aded5bc..edc7e5e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,7 +69,7 @@ under the License.
<junit.version>4.12</junit.version>
<flink.forkCount>1C</flink.forkCount>
<flink.reuseForks>true</flink.reuseForks>
- <flink.version>1.12.1</flink.version>
+ <flink.version>1.14.0</flink.version>
<zookeeper.version>3.4.14</zookeeper.version>
<!-- Can be set to any value to reproduce a specific build. -->