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 fe2966d [FLINK-29591] Add built-in UDFs to convert between arrays and
vectors
fe2966d is described below
commit fe2966db6b5419459c98752cd1eaced9d4eb46ea
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Tue Oct 18 10:55:26 2022 +0800
[FLINK-29591] Add built-in UDFs to convert between arrays and vectors
This closes #163.
---
docs/content/docs/operators/functions.md | 236 +++++++++++++++++++++
.../flink/ml/examples/ArrayToVectorExample.java | 58 +++++
.../flink/ml/examples/VectorToArrayExample.java | 64 ++++++
.../main/java/org/apache/flink/ml/Functions.java | 107 ++++++++++
.../java/org/apache/flink/ml/FunctionsTest.java | 132 ++++++++++++
.../pyflink/examples/ml/arraytovector_example.py | 60 ++++++
.../pyflink/examples/ml/vectortoarray_example.py | 62 ++++++
flink-ml-python/pyflink/ml/core/wrapper.py | 11 +-
flink-ml-python/pyflink/ml/lib/functions.py | 34 +++
.../pyflink/ml/lib/tests/test_functions.py | 114 ++++++++++
10 files changed, 877 insertions(+), 1 deletion(-)
diff --git a/docs/content/docs/operators/functions.md
b/docs/content/docs/operators/functions.md
new file mode 100644
index 0000000..0cddbab
--- /dev/null
+++ b/docs/content/docs/operators/functions.md
@@ -0,0 +1,236 @@
+---
+title: "Functions"
+type: docs
+weight: 2
+aliases:
+- /operators/functions.html
+---
+<!--
+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.
+-->
+
+## Functions
+
+Flink ML provides users with some built-in table functions for data
+transformations. This page gives a brief overview of them.
+
+### vectorToArray
+
+This function converts a column of Flink ML sparse/dense vectors into a column
+of double arrays.
+
+{{< tabs vectorToArray_examples >}}
+
+{{< tab "Java">}}
+```java
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.linalg.typeinfo.VectorTypeInfo;
+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.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.ml.Functions.vectorToArray;
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Simple program that converts a column of dense/sparse vectors into a
column of double arrays. */
+public class VectorToArrayExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input vector data.
+ List<Vector> vectors =
+ Arrays.asList(
+ Vectors.dense(0.0, 0.0),
+ Vectors.sparse(2, new int[] {1}, new double[] {1.0}));
+ Table inputTable =
+ tEnv.fromDataStream(env.fromCollection(vectors,
VectorTypeInfo.INSTANCE))
+ .as("vector");
+
+ // Converts each vector to a double array.
+ Table outputTable = inputTable.select($("vector"),
vectorToArray($("vector")).as("array"));
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+ Vector vector = row.getFieldAs("vector");
+ Double[] doubleArray = row.getFieldAs("array");
+ System.out.printf(
+ "Input vector: %s\tOutput double array: %s\n",
+ vector, Arrays.toString(doubleArray));
+ }
+ }
+}
+```
+{{< /tab>}}
+
+{{< tab "Python">}}
+```python
+# Simple program that converts a column of dense/sparse vectors into a column
of double arrays.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.table import StreamTableEnvironment
+
+from pyflink.ml.core.linalg import Vectors, VectorTypeInfo
+
+from pyflink.ml.lib.functions import vector_to_array
+from pyflink.table.expressions import col
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input vector data
+vectors = [
+ (Vectors.dense(0.0, 0.0),),
+ (Vectors.sparse(2, [1], [1.0]),),
+]
+input_table = t_env.from_data_stream(
+ env.from_collection(
+ vectors,
+ type_info=Types.ROW_NAMED(
+ ['vector'],
+ [VectorTypeInfo()])
+ ))
+
+# convert each vector to a double array
+output_table =
input_table.select(vector_to_array(col('vector')).alias('array'))
+
+# extract and display the results
+output_values = [x for x in
+ t_env.to_data_stream(output_table).map(lambda r:
r).execute_and_collect()]
+
+output_values.sort(key=lambda x: x[0])
+
+field_names = output_table.get_schema().get_field_names()
+for i in range(len(output_values)):
+ vector = vectors[i][0]
+ double_array = output_values[i][field_names.index("array")]
+ print("Input vector: %s \t output double array: %s" % (vector,
double_array))
+```
+{{< /tab>}}
+
+{{< /tabs>}}
+
+### arrayToVector
+
+This function converts a column of arrays of numeric type into a column of
+DenseVector instances.
+
+{{< tabs arrayToVector_examples >}}
+
+{{< tab "Java">}}
+```java
+import org.apache.flink.ml.linalg.Vector;
+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.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.ml.Functions.arrayToVector;
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Simple program that converts a column of double arrays into a column of
dense vectors. */
+public class ArrayToVectorExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input double array data.
+ List<double[]> doubleArrays =
+ Arrays.asList(new double[] {0.0, 0.0}, new double[] {0.0,
1.0});
+ Table inputTable =
tEnv.fromDataStream(env.fromCollection(doubleArrays)).as("array");
+
+ // Converts each double array to a dense vector.
+ Table outputTable = inputTable.select($("array"),
arrayToVector($("array")).as("vector"));
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+ Double[] doubleArray = row.getFieldAs("array");
+ Vector vector = row.getFieldAs("vector");
+ System.out.printf(
+ "Input double array: %s\tOutput vector: %s\n",
+ Arrays.toString(doubleArray), vector);
+ }
+ }
+}
+```
+{{< /tab>}}
+
+{{< tab "Python">}}
+```python
+# Simple program that converts a column of double arrays into a column of
dense vectors.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.lib.functions import array_to_vector
+from pyflink.table import StreamTableEnvironment
+from pyflink.table.expressions import col
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input double array data
+double_arrays = [
+ ([0.0, 0.0],),
+ ([0.0, 1.0],),
+]
+input_table = t_env.from_data_stream(
+ env.from_collection(
+ double_arrays,
+ type_info=Types.ROW_NAMED(
+ ['array'],
+ [Types.PRIMITIVE_ARRAY(Types.DOUBLE())])
+ ))
+
+# convert each double array to a dense vector
+output_table =
input_table.select(array_to_vector(col('array')).alias('vector'))
+
+# extract and display the results
+field_names = output_table.get_schema().get_field_names()
+
+output_values = [x[field_names.index('vector')] for x in
+ t_env.to_data_stream(output_table).execute_and_collect()]
+
+output_values.sort(key=lambda x: x.get(1))
+
+for i in range(len(output_values)):
+ double_array = double_arrays[i][0]
+ vector = output_values[i]
+ print("Input double array: %s \t output vector: %s" % (double_array,
vector))
+```
+{{< /tab>}}
+
+{{< /tabs>}}
diff --git
a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/ArrayToVectorExample.java
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/ArrayToVectorExample.java
new file mode 100644
index 0000000..85c57b7
--- /dev/null
+++
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/ArrayToVectorExample.java
@@ -0,0 +1,58 @@
+/*
+ * 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.examples;
+
+import org.apache.flink.ml.linalg.Vector;
+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.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.ml.Functions.arrayToVector;
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Simple program that converts a column of double arrays into a column of
dense vectors. */
+public class ArrayToVectorExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input double array data.
+ List<double[]> doubleArrays =
+ Arrays.asList(new double[] {0.0, 0.0}, new double[] {0.0,
1.0});
+ Table inputTable =
tEnv.fromDataStream(env.fromCollection(doubleArrays)).as("array");
+
+ // Converts each double array to a dense vector.
+ Table outputTable = inputTable.select($("array"),
arrayToVector($("array")).as("vector"));
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+ Double[] doubleArray = row.getFieldAs("array");
+ Vector vector = row.getFieldAs("vector");
+ System.out.printf(
+ "Input double array: %s\tOutput vector: %s\n",
+ Arrays.toString(doubleArray), vector);
+ }
+ }
+}
diff --git
a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/VectorToArrayExample.java
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/VectorToArrayExample.java
new file mode 100644
index 0000000..733fe45
--- /dev/null
+++
b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/VectorToArrayExample.java
@@ -0,0 +1,64 @@
+/*
+ * 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.examples;
+
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.linalg.typeinfo.VectorTypeInfo;
+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.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.ml.Functions.vectorToArray;
+import static org.apache.flink.table.api.Expressions.$;
+
+/** Simple program that converts a column of dense/sparse vectors into a
column of double arrays. */
+public class VectorToArrayExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input vector data.
+ List<Vector> vectors =
+ Arrays.asList(
+ Vectors.dense(0.0, 0.0),
+ Vectors.sparse(2, new int[] {1}, new double[] {1.0}));
+ Table inputTable =
+ tEnv.fromDataStream(env.fromCollection(vectors,
VectorTypeInfo.INSTANCE))
+ .as("vector");
+
+ // Converts each vector to a double array.
+ Table outputTable = inputTable.select($("vector"),
vectorToArray($("vector")).as("array"));
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+ Vector vector = row.getFieldAs("vector");
+ Double[] doubleArray = row.getFieldAs("array");
+ System.out.printf(
+ "Input vector: %s\tOutput double array: %s\n",
+ vector, Arrays.toString(doubleArray));
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/Functions.java
b/flink-ml-lib/src/main/java/org/apache/flink/ml/Functions.java
new file mode 100644
index 0000000..694b35e
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/Functions.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml;
+
+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.table.api.ApiExpression;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.types.inference.TypeInference;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.util.Optional;
+
+import static org.apache.flink.table.api.Expressions.call;
+
+/** Built-in table functions for data transformations. */
+@SuppressWarnings("unused")
+public class Functions {
+ /** Converts a column of {@link Vector}s into a column of double arrays. */
+ public static ApiExpression vectorToArray(Object... arguments) {
+ return call(VectorToArrayFunction.class, arguments);
+ }
+
+ /**
+ * A {@link ScalarFunction} that converts a column of {@link Vector}s into
a column of double
+ * arrays.
+ */
+ public static class VectorToArrayFunction extends ScalarFunction {
+ public double[] eval(Vector vector) {
+ return vector.toArray();
+ }
+
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ return TypeInference.newBuilder()
+ .outputTypeStrategy(
+ callContext ->
+ Optional.of(
+ DataTypes.ARRAY(
+ DataTypes.DOUBLE()
+ .notNull()
+
.bridgedTo(double.class))))
+ .build();
+ }
+ }
+
+ /**
+ * Converts a column of arrays of numeric type into a column of {@link
DenseVector} instances.
+ */
+ public static ApiExpression arrayToVector(Object... arguments) {
+ return call(ArrayToVectorFunction.class, arguments);
+ }
+
+ /**
+ * A {@link ScalarFunction} that converts a column of arrays of numeric
type into a column of
+ * {@link DenseVector} instances.
+ */
+ public static class ArrayToVectorFunction extends ScalarFunction {
+ public DenseVector eval(double[] array) {
+ return Vectors.dense(array);
+ }
+
+ public DenseVector eval(Double[] array) {
+ return eval(ArrayUtils.toPrimitive(array));
+ }
+
+ public DenseVector eval(Number[] array) {
+ double[] doubles = new double[array.length];
+ for (int i = 0; i < array.length; i++) {
+ doubles[i] = array[i].doubleValue();
+ }
+ return eval(doubles);
+ }
+
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ return TypeInference.newBuilder()
+ .outputTypeStrategy(
+ callContext ->
+ Optional.of(
+
DataTypes.of(DenseVectorTypeInfo.INSTANCE)
+ .toDataType(typeFactory)))
+ .build();
+ }
+ }
+}
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/FunctionsTest.java
b/flink-ml-lib/src/test/java/org/apache/flink/ml/FunctionsTest.java
new file mode 100644
index 0000000..3346764
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/FunctionsTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.VectorTypeInfo;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.ArrayUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.flink.ml.Functions.arrayToVector;
+import static org.apache.flink.ml.Functions.vectorToArray;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+/** Tests {@link Functions}. */
+public class FunctionsTest extends AbstractTestBase {
+ private static final List<double[]> doubleArrays =
+ Arrays.asList(new double[] {0.0, 0.0}, new double[] {0.0, 1.0});
+
+ private static final List<float[]> floatArrays =
+ Arrays.asList(new float[] {0.0f, 0.0f}, new float[] {0.0f, 1.0f});
+
+ private static final List<int[]> intArrays = Arrays.asList(new int[] {0,
0}, new int[] {0, 1});
+
+ private static final List<long[]> longArrays =
+ Arrays.asList(new long[] {0, 0}, new long[] {0, 1});
+
+ private static final List<DenseVector> denseVectors =
+ Arrays.asList(Vectors.dense(0.0, 0.0), Vectors.dense(0.0, 1.0));
+
+ private static final List<SparseVector> sparseVectors =
+ Arrays.asList(
+ Vectors.sparse(2, new int[0], new double[0]),
+ Vectors.sparse(2, new int[] {1}, new double[] {1.0}));
+
+ private static final List<Vector> mixedVectors =
+ Arrays.asList(
+ Vectors.dense(0.0, 0.0), Vectors.sparse(2, new int[] {1},
new double[] {1.0}));
+
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ @Before
+ public void before() {
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env);
+ }
+
+ @Test
+ public void testVectorToArray() {
+ testVectorToArray(denseVectors, null);
+ testVectorToArray(sparseVectors, null);
+ testVectorToArray(mixedVectors, VectorTypeInfo.INSTANCE);
+ }
+
+ private <T> void testVectorToArray(
+ List<T> vectors, @Nullable TypeInformation<T>
vectorTypeInformation) {
+ Table inputTable;
+ if (vectorTypeInformation == null) {
+ inputTable = tEnv.fromDataStream(env.fromCollection(vectors));
+ } else {
+ inputTable = tEnv.fromDataStream(env.fromCollection(vectors,
vectorTypeInformation));
+ }
+ inputTable = inputTable.as("vector");
+
+ Table outputTable =
inputTable.select(vectorToArray($("vector")).as("array"));
+
+ List<Row> outputValues =
IteratorUtils.toList(outputTable.execute().collect());
+
+ assertEquals(outputValues.size(), doubleArrays.size());
+ for (int i = 0; i < doubleArrays.size(); i++) {
+ Double[] doubles = outputValues.get(i).getFieldAs("array");
+ assertArrayEquals(doubleArrays.get(i),
ArrayUtils.toPrimitive(doubles));
+ }
+ }
+
+ @Test
+ public void testArrayToVector() {
+ testArrayToVector(doubleArrays);
+ testArrayToVector(floatArrays);
+ testArrayToVector(intArrays);
+ testArrayToVector(longArrays);
+ }
+
+ private <T> void testArrayToVector(List<T> array) {
+ Table inputTable =
tEnv.fromDataStream(env.fromCollection(array)).as("array");
+
+ Table outputTable =
inputTable.select(arrayToVector($("array")).as("vector"));
+
+ List<Row> outputValues =
IteratorUtils.toList(outputTable.execute().collect());
+
+ assertEquals(outputValues.size(), denseVectors.size());
+ for (int i = 0; i < denseVectors.size(); i++) {
+ DenseVector vector = outputValues.get(i).getFieldAs("vector");
+ assertEquals(denseVectors.get(i), vector);
+ }
+ }
+}
diff --git a/flink-ml-python/pyflink/examples/ml/arraytovector_example.py
b/flink-ml-python/pyflink/examples/ml/arraytovector_example.py
new file mode 100644
index 0000000..78ad427
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/arraytovector_example.py
@@ -0,0 +1,60 @@
+################################################################################
+# 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.
+################################################################################
+
+# Simple program that converts a column of double arrays into a column of
dense vectors.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.lib.functions import array_to_vector
+from pyflink.table import StreamTableEnvironment
+from pyflink.table.expressions import col
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input double array data
+double_arrays = [
+ ([0.0, 0.0],),
+ ([0.0, 1.0],),
+]
+input_table = t_env.from_data_stream(
+ env.from_collection(
+ double_arrays,
+ type_info=Types.ROW_NAMED(
+ ['array'],
+ [Types.PRIMITIVE_ARRAY(Types.DOUBLE())])
+ ))
+
+# convert each double array to a dense vector
+output_table =
input_table.select(array_to_vector(col('array')).alias('vector'))
+
+# extract and display the results
+field_names = output_table.get_schema().get_field_names()
+
+output_values = [x[field_names.index('vector')] for x in
+ t_env.to_data_stream(output_table).execute_and_collect()]
+
+output_values.sort(key=lambda x: x.get(1))
+
+for i in range(len(output_values)):
+ double_array = double_arrays[i][0]
+ vector = output_values[i]
+ print("Input double array: %s \t output vector: %s" % (double_array,
vector))
diff --git a/flink-ml-python/pyflink/examples/ml/vectortoarray_example.py
b/flink-ml-python/pyflink/examples/ml/vectortoarray_example.py
new file mode 100644
index 0000000..0ead802
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/vectortoarray_example.py
@@ -0,0 +1,62 @@
+################################################################################
+# 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.
+################################################################################
+
+# Simple program that converts a column of dense/sparse vectors into a column
of double arrays.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.table import StreamTableEnvironment
+
+from pyflink.ml.core.linalg import Vectors, VectorTypeInfo
+
+from pyflink.ml.lib.functions import vector_to_array
+from pyflink.table.expressions import col
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input vector data
+vectors = [
+ (Vectors.dense(0.0, 0.0),),
+ (Vectors.sparse(2, [1], [1.0]),),
+]
+input_table = t_env.from_data_stream(
+ env.from_collection(
+ vectors,
+ type_info=Types.ROW_NAMED(
+ ['vector'],
+ [VectorTypeInfo()])
+ ))
+
+# convert each vector to a double array
+output_table =
input_table.select(vector_to_array(col('vector')).alias('array'))
+
+# extract and display the results
+output_values = [x for x in
+ t_env.to_data_stream(output_table).map(lambda r:
r).execute_and_collect()]
+
+output_values.sort(key=lambda x: x[0])
+
+field_names = output_table.get_schema().get_field_names()
+for i in range(len(output_values)):
+ vector = vectors[i][0]
+ double_array = output_values[i][field_names.index("array")]
+ print("Input vector: %s \t output double array: %s" % (vector,
double_array))
diff --git a/flink-ml-python/pyflink/ml/core/wrapper.py
b/flink-ml-python/pyflink/ml/core/wrapper.py
index 191d8de..e965c0c 100644
--- a/flink-ml-python/pyflink/ml/core/wrapper.py
+++ b/flink-ml-python/pyflink/ml/core/wrapper.py
@@ -22,7 +22,7 @@ from py4j.java_gateway import JavaObject, get_java_class
from pyflink.common import typeinfo, Time
from pyflink.common.typeinfo import _from_java_type, TypeInformation,
_is_instance_of
from pyflink.java_gateway import get_gateway
-from pyflink.table import Table, StreamTableEnvironment
+from pyflink.table import Table, StreamTableEnvironment, Expression
from pyflink.util.java_utils import to_jarray
from pyflink.ml.core.api import Model, Transformer, AlgoOperator, Stage,
Estimator
@@ -360,3 +360,12 @@ def _to_java_tables(*inputs: Table):
"""
gateway = get_gateway()
return to_jarray(gateway.jvm.org.apache.flink.table.api.Table, [t._j_table
for t in inputs])
+
+
+def call_java_table_function(java_table_function_name: str, *args):
+ _function = get_gateway().jvm
+ for member_name in java_table_function_name.split('.'):
+ _function = _function.__getattr__(member_name)
+ return Expression(_function(to_jarray(
+ get_gateway().jvm.java.lang.Object,
+ [expression._j_expr for expression in args])))
diff --git a/flink-ml-python/pyflink/ml/lib/functions.py
b/flink-ml-python/pyflink/ml/lib/functions.py
new file mode 100644
index 0000000..1468834
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/functions.py
@@ -0,0 +1,34 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.ml.core.wrapper import call_java_table_function
+from pyflink.table import Expression
+
+
+def vector_to_array(*args) -> Expression:
+ """
+ Converts a column of :class:`Vector`s into a column of double arrays.
+ """
+ return
call_java_table_function('org.apache.flink.ml.Functions.vectorToArray', *args)
+
+
+def array_to_vector(*args) -> Expression:
+ """
+ Converts a column of arrays of numeric type into a column of
+ :class:`DenseVector` instances.
+ """
+ return
call_java_table_function('org.apache.flink.ml.Functions.arrayToVector', *args)
diff --git a/flink-ml-python/pyflink/ml/lib/tests/test_functions.py
b/flink-ml-python/pyflink/ml/lib/tests/test_functions.py
new file mode 100644
index 0000000..6ad059b
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/tests/test_functions.py
@@ -0,0 +1,114 @@
+################################################################################
+# 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.
+################################################################################
+from pyflink.common import Types
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo,
SparseVectorTypeInfo, \
+ VectorTypeInfo
+from pyflink.ml.lib.functions import vector_to_array, array_to_vector
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+from pyflink.table.expressions import col
+
+
+class FunctionsTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(FunctionsTest, self).setUp()
+
+ self.double_arrays = [
+ ([0.0, 0.0],),
+ ([0.0, 1.0],),
+ ]
+
+ self.float_arrays = [
+ ([float(0.0), float(0.0)],),
+ ([float(0.0), float(1.0)],),
+ ]
+
+ self.int_arrays = [
+ ([0, 0],),
+ ([0, 1],),
+ ]
+
+ self.dense_vectors = [
+ (Vectors.dense(0.0, 0.0),),
+ (Vectors.dense(0.0, 1.0),),
+ ]
+
+ self.sparse_vectors = [
+ (Vectors.sparse(2, [], []),),
+ (Vectors.sparse(2, [1], [1.0]),),
+ ]
+
+ self.mixed_vectors = [
+ (Vectors.dense(0.0, 0.0),),
+ (Vectors.sparse(2, [1], [1.0]),),
+ ]
+
+ def test_vector_to_array(self):
+ self._test_vector_to_array(self.dense_vectors, DenseVectorTypeInfo())
+ self._test_vector_to_array(self.sparse_vectors, SparseVectorTypeInfo())
+ self._test_vector_to_array(self.mixed_vectors, VectorTypeInfo())
+
+ def _test_vector_to_array(self, vectors, vector_type_info):
+ input_table = self.t_env.from_data_stream(
+ self.env.from_collection(vectors,
+ type_info=Types.ROW_NAMED(
+ ['vector'],
+ [vector_type_info])
+ ))
+
+ output_table =
input_table.select(vector_to_array(col('vector')).alias('array'))
+
+ output_values = [x['array'] for x in
self.t_env.to_data_stream(output_table)
+ .map(lambda r: r).execute_and_collect()]
+
+ self.assertEqual(len(output_values), len(self.double_arrays))
+
+ output_values.sort(key=lambda x: x[1])
+
+ for i in range(len(self.double_arrays)):
+ self.assertEqual(self.double_arrays[i][0], output_values[i])
+
+ def test_array_to_vector(self):
+ self._test_array_to_vector(self.double_arrays, Types.DOUBLE())
+ self._test_array_to_vector(self.float_arrays, Types.FLOAT())
+ self._test_array_to_vector(self.int_arrays, Types.INT())
+ self._test_array_to_vector(self.int_arrays, Types.LONG())
+
+ def _test_array_to_vector(self, arrays, array_element_type_info):
+ input_table = self.t_env.from_data_stream(
+ self.env.from_collection(
+ arrays,
+ type_info=Types.ROW_NAMED(
+ ['array'],
+ [Types.PRIMITIVE_ARRAY(array_element_type_info)]
+ )
+ )
+ )
+
+ output_table =
input_table.select(array_to_vector(col('array')).alias('vector'))
+
+ field_names = output_table.get_schema().get_field_names()
+
+ output_values = [x[field_names.index('vector')] for x in
+
self.t_env.to_data_stream(output_table).execute_and_collect()]
+
+ self.assertEqual(len(output_values), len(self.dense_vectors))
+
+ output_values.sort(key=lambda x: x.get(1))
+
+ for i in range(len(self.dense_vectors)):
+ self.assertEqual(self.dense_vectors[i][0], output_values[i])
