[FLINK-33118] Remove the PythonBridgeUtils
This closes #256.
diff --git a/flink-ml-dist/pom.xml b/flink-ml-dist/pom.xml
index 7c260ec..1bb239d 100644
--- a/flink-ml-dist/pom.xml
+++ b/flink-ml-dist/pom.xml
@@ -46,12 +46,6 @@
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-ml-python-${flink.main.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
<!-- Stateful Functions Dependencies -->
<dependency>
diff --git a/flink-ml-dist/src/main/assemblies/bin.xml b/flink-ml-dist/src/main/assemblies/bin.xml
index 00ad9dc..d260e0d 100644
--- a/flink-ml-dist/src/main/assemblies/bin.xml
+++ b/flink-ml-dist/src/main/assemblies/bin.xml
@@ -37,7 +37,6 @@
<include>org.apache.flink:statefun-flink-core</include>
<include>org.apache.flink:flink-ml-uber-${flink.main.version}</include>
<include>org.apache.flink:flink-ml-examples-${flink.main.version}</include>
- <include>org.apache.flink:flink-ml-python-${flink.main.version}</include>
</includes>
</dependencySet>
</dependencySets>
diff --git a/flink-ml-python/pom.xml b/flink-ml-python/pom.xml
index 34e7e5a..c481b21 100644
--- a/flink-ml-python/pom.xml
+++ b/flink-ml-python/pom.xml
@@ -32,20 +32,6 @@
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>${flink.python.artifact}</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
<version>${flink.version}</version>
<type>test-jar</type>
diff --git a/flink-ml-python/pyflink/ml/feature/tests/test_indextostringmodel.py b/flink-ml-python/pyflink/ml/feature/tests/test_indextostringmodel.py
index cae0673..34779b6 100644
--- a/flink-ml-python/pyflink/ml/feature/tests/test_indextostringmodel.py
+++ b/flink-ml-python/pyflink/ml/feature/tests/test_indextostringmodel.py
@@ -45,8 +45,8 @@
))
self.expected_prediction = [
- Row(0, 3, 'a', '2.0'),
- Row(1, 2, 'b', '1.0'),
+ Row(input_col1=0, input_col2=3, output_col1='a', output_col2='2.0'),
+ Row(input_col1=1, input_col2=2, output_col1='b', output_col2='1.0'),
]
def test_output_schema(self):
diff --git a/flink-ml-python/pyflink/ml/feature/tests/test_minhashlsh.py b/flink-ml-python/pyflink/ml/feature/tests/test_minhashlsh.py
index 9fed97c..a345c81 100644
--- a/flink-ml-python/pyflink/ml/feature/tests/test_minhashlsh.py
+++ b/flink-ml-python/pyflink/ml/feature/tests/test_minhashlsh.py
@@ -179,8 +179,8 @@
.set_num_hash_tables(5) \
.set_num_hash_functions_per_table(1)
expected = [
- Row(0, 0.75),
- Row(1, 0.75),
+ Row(id=0, distCol=0.75),
+ Row(id=1, distCol=0.75),
]
model: MinHashLSHModel = lsh.fit(self.data)
@@ -198,8 +198,8 @@
.set_num_hash_tables(5) \
.set_num_hash_functions_per_table(1)
expected = [
- Row(0, 0.75),
- Row(1, 0.75),
+ Row(id=0, distCol=0.75),
+ Row(id=1, distCol=0.75),
]
model: MinHashLSHModel = lsh.fit(self.data)
@@ -230,6 +230,8 @@
Row(1, 5, .5),
Row(2, 5, .5)
]
+ for r in expected:
+ r.set_field_names(['datasetA.id', 'datasetB.id', 'distCol'])
output = model.approx_similarity_join(data_a, data_b, .6, "id")
actual_result = [r for r in self.t_env.to_data_stream(output).execute_and_collect()]
diff --git a/flink-ml-python/pyflink/ml/feature/tests/test_sqltransformer.py b/flink-ml-python/pyflink/ml/feature/tests/test_sqltransformer.py
index 85d05a4..52ff82b 100644
--- a/flink-ml-python/pyflink/ml/feature/tests/test_sqltransformer.py
+++ b/flink-ml-python/pyflink/ml/feature/tests/test_sqltransformer.py
@@ -34,8 +34,8 @@
['id', 'v1', 'v2'],
[Types.INT(), Types.DOUBLE(), Types.DOUBLE()])))
self.expected_output = [
- (0, 1.0, 3.0, 4.0, 3.0),
- (2, 2.0, 5.0, 7.0, 10.0)
+ Row(id=0, v1=1.0, v2=3.0, v3=4.0, v4=3.0),
+ Row(id=2, v1=2.0, v2=5.0, v3=7.0, v4=10.0)
]
def test_param(self):
@@ -62,4 +62,4 @@
actual_output.sort(key=lambda x: x[0])
self.assertEqual(len(self.expected_output), len(actual_output))
for i in range(len(actual_output)):
- self.assertEqual(Row(*self.expected_output[i]), actual_output[i])
+ self.assertEqual(self.expected_output[i], actual_output[i])
diff --git a/flink-ml-python/pyflink/ml/feature/tests/test_stringindexer.py b/flink-ml-python/pyflink/ml/feature/tests/test_stringindexer.py
index aa61503..0809533 100644
--- a/flink-ml-python/pyflink/ml/feature/tests/test_stringindexer.py
+++ b/flink-ml-python/pyflink/ml/feature/tests/test_stringindexer.py
@@ -58,11 +58,11 @@
[Types.STRING(), Types.DOUBLE()])))
self.expected_alphabetic_asc_predict_data = [
- Row('a', 2.0, 0, 3),
- Row('b', 1.0, 1, 2),
- Row('e', 2.0, 4, 3),
- Row('f', None, 4, 4),
- Row(None, None, 4, 4),
+ Row(input_col1='a', input_col2=2.0, output_col1=0, output_col2=3),
+ Row(input_col1='b', input_col2=1.0, output_col1=1, output_col2=2),
+ Row(input_col1='e', input_col2=2.0, output_col1=4, output_col2=3),
+ Row(input_col1='f', input_col2=None, output_col1=4, output_col2=4),
+ Row(input_col1=None, input_col2=None, output_col1=4, output_col2=4),
]
def test_param(self):
@@ -122,11 +122,11 @@
.set_string_order_type("frequencyDesc")
expected_predict_data = [
- Row('a', 2.0, 1, 0),
- Row('b', 1.0, 0, 2),
- Row('e', 2.0, 2, 0),
- Row('f', None, 2, 2),
- Row(None, None, 2, 2),
+ Row(input_col1='a', input_col2=2.0, output_col1=1, output_col2=0),
+ Row(input_col1='b', input_col2=1.0, output_col1=0, output_col2=2),
+ Row(input_col1='e', input_col2=2.0, output_col1=2, output_col2=0),
+ Row(input_col1='f', input_col2=None, output_col1=2, output_col2=2),
+ Row(input_col1=None, input_col2=None, output_col1=2, output_col2=2),
]
output = string_indexer.fit(self.train_table).transform(self.predict_table)[0]
diff --git a/flink-ml-python/pyflink/ml/stats/tests/test_chisqtest.py b/flink-ml-python/pyflink/ml/stats/tests/test_chisqtest.py
index 290e468..1c4edc5 100644
--- a/flink-ml-python/pyflink/ml/stats/tests/test_chisqtest.py
+++ b/flink-ml-python/pyflink/ml/stats/tests/test_chisqtest.py
@@ -50,8 +50,8 @@
)
self.expected_output_data = [
- Row(0, 0.03419350755, 6, 13.61904761905),
- Row(1, 0.24220177737, 6, 7.94444444444)]
+ Row(featureIndex=0, pValue=0.03419350755, degreeOfFreedom=6, statistic=13.61904761905),
+ Row(featureIndex=1, pValue=0.24220177737, degreeOfFreedom=6, statistic=7.94444444444)]
def test_param(self):
chi_sq_test = ChiSqTest()
diff --git a/flink-ml-python/pyflink/ml/wrapper.py b/flink-ml-python/pyflink/ml/wrapper.py
index f30f218..50f0105 100644
--- a/flink-ml-python/pyflink/ml/wrapper.py
+++ b/flink-ml-python/pyflink/ml/wrapper.py
@@ -15,16 +15,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-import pickle
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from py4j.java_gateway import JavaObject, get_java_class
-from pyflink.common import typeinfo, Time, Row, RowKind
-from pyflink.common.typeinfo import _from_java_type, TypeInformation, _is_instance_of, Types, \
- ExternalTypeInfo, RowTypeInfo, TupleTypeInfo
-from pyflink.datastream import utils
-from pyflink.datastream.utils import pickled_bytes_to_python_obj
+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, Expression
from pyflink.util.java_utils import to_jarray
@@ -60,36 +56,6 @@
typeinfo._from_java_type = _from_java_type_wrapper
-# TODO: Remove this class after Flink ML depends on a Flink version
-# with FLINK-30168 and FLINK-29477 fixed.
-def convert_to_python_obj_wrapper(data, type_info):
- if type_info == Types.PICKLED_BYTE_ARRAY():
- return pickle.loads(data)
- elif isinstance(type_info, ExternalTypeInfo):
- return convert_to_python_obj_wrapper(data, type_info._type_info)
- else:
- gateway = get_gateway()
- pickle_bytes = gateway.jvm.org.apache.flink.ml.python.PythonBridgeUtils. \
- getPickledBytesFromJavaObject(data, type_info.get_java_type_info())
- if isinstance(type_info, RowTypeInfo) or isinstance(type_info, TupleTypeInfo):
- field_data = zip(list(pickle_bytes[1:]), type_info.get_field_types())
- fields = []
- for data, field_type in field_data:
- if len(data) == 0:
- fields.append(None)
- else:
- fields.append(pickled_bytes_to_python_obj(data, field_type))
- if isinstance(type_info, RowTypeInfo):
- return Row.of_kind(RowKind(int.from_bytes(pickle_bytes[0], 'little')), *fields)
- else:
- return tuple(fields)
- else:
- return pickled_bytes_to_python_obj(pickle_bytes, type_info)
-
-
-utils.convert_to_python_obj = convert_to_python_obj_wrapper
-
-
class JavaWrapper(ABC):
"""
Wrapper class for a Java object.
diff --git a/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java b/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java
deleted file mode 100644
index 36eca36..0000000
--- a/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java
+++ /dev/null
@@ -1,226 +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.python;
-
-import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
-import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
-import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.typeutils.ListTypeInfo;
-import org.apache.flink.api.java.typeutils.MapTypeInfo;
-import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
-import org.apache.flink.api.python.shaded.net.razorvine.pickle.Pickler;
-import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
-import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
-import org.apache.flink.streaming.api.typeinfo.python.PickledByteArrayTypeInfo;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Preconditions;
-
-import java.io.IOException;
-import java.sql.Date;
-import java.sql.Time;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.FLOAT_TYPE_INFO;
-import static org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.DATE;
-import static org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.TIME;
-
-/**
- * Utility functions used to override PyFlink methods to provide a temporary solution for certain
- * bugs.
- */
-// TODO: Remove this class after Flink ML depends on a Flink version with FLINK-30168 and
-// FLINK-29477 fixed.
-public class PythonBridgeUtils {
- public static Object getPickledBytesFromJavaObject(Object obj, TypeInformation<?> dataType)
- throws IOException {
- Pickler pickler = new Pickler();
-
- // triggers the initialization process
- org.apache.flink.api.common.python.PythonBridgeUtils.getPickledBytesFromJavaObject(
- null, null);
-
- if (obj == null) {
- return pickler.dumps(null);
- } else {
- if (dataType instanceof SqlTimeTypeInfo) {
- SqlTimeTypeInfo<?> sqlTimeTypeInfo =
- SqlTimeTypeInfo.getInfoFor(dataType.getTypeClass());
- if (sqlTimeTypeInfo == DATE) {
- return pickler.dumps(((Date) obj).toLocalDate().toEpochDay());
- } else if (sqlTimeTypeInfo == TIME) {
- return pickler.dumps(((Time) obj).toLocalTime().toNanoOfDay() / 1000);
- }
- } else if (dataType instanceof RowTypeInfo || dataType instanceof TupleTypeInfo) {
- TypeInformation<?>[] fieldTypes = ((TupleTypeInfoBase<?>) dataType).getFieldTypes();
- int arity =
- dataType instanceof RowTypeInfo
- ? ((Row) obj).getArity()
- : ((Tuple) obj).getArity();
-
- List<Object> fieldBytes = new ArrayList<>(arity + 1);
- if (dataType instanceof RowTypeInfo) {
- fieldBytes.add(new byte[] {((Row) obj).getKind().toByteValue()});
- }
- for (int i = 0; i < arity; i++) {
- Object field =
- dataType instanceof RowTypeInfo
- ? ((Row) obj).getField(i)
- : ((Tuple) obj).getField(i);
- fieldBytes.add(getPickledBytesFromJavaObject(field, fieldTypes[i]));
- }
- return fieldBytes;
- } else if (dataType instanceof BasicArrayTypeInfo
- || dataType instanceof PrimitiveArrayTypeInfo
- || dataType instanceof ObjectArrayTypeInfo) {
- Object[] objects;
- TypeInformation<?> elementType;
- if (dataType instanceof BasicArrayTypeInfo) {
- objects = (Object[]) obj;
- elementType = ((BasicArrayTypeInfo<?, ?>) dataType).getComponentInfo();
- } else if (dataType instanceof PrimitiveArrayTypeInfo) {
- objects = primitiveArrayConverter(obj, dataType);
- elementType = ((PrimitiveArrayTypeInfo<?>) dataType).getComponentType();
- } else {
- objects = (Object[]) obj;
- elementType = ((ObjectArrayTypeInfo<?, ?>) dataType).getComponentInfo();
- }
-
- List<Object> serializedElements = new ArrayList<>(objects.length);
-
- for (Object object : objects) {
- serializedElements.add(getPickledBytesFromJavaObject(object, elementType));
- }
- return pickler.dumps(serializedElements);
- } else if (dataType instanceof MapTypeInfo) {
- List<List<Object>> serializedMapKV = new ArrayList<>(2);
- Map<Object, Object> mapObj = (Map) obj;
- List<Object> keyBytesList = new ArrayList<>(mapObj.size());
- List<Object> valueBytesList = new ArrayList<>(mapObj.size());
- for (Map.Entry entry : mapObj.entrySet()) {
- keyBytesList.add(
- getPickledBytesFromJavaObject(
- entry.getKey(), ((MapTypeInfo) dataType).getKeyTypeInfo()));
- valueBytesList.add(
- getPickledBytesFromJavaObject(
- entry.getValue(), ((MapTypeInfo) dataType).getValueTypeInfo()));
- }
- serializedMapKV.add(keyBytesList);
- serializedMapKV.add(valueBytesList);
- return pickler.dumps(serializedMapKV);
- } else if (dataType instanceof ListTypeInfo) {
- List objects = (List) obj;
- List<Object> serializedElements = new ArrayList<>(objects.size());
- TypeInformation elementType = ((ListTypeInfo) dataType).getElementTypeInfo();
- for (Object object : objects) {
- serializedElements.add(getPickledBytesFromJavaObject(object, elementType));
- }
- return pickler.dumps(serializedElements);
- }
- if (dataType instanceof BasicTypeInfo
- && BasicTypeInfo.getInfoFor(dataType.getTypeClass()) == FLOAT_TYPE_INFO) {
- // Serialization of float type with pickler loses precision.
- return pickler.dumps(String.valueOf(obj));
- } else if (dataType instanceof PickledByteArrayTypeInfo
- || dataType instanceof BasicTypeInfo) {
- return pickler.dumps(obj);
- } else {
- // other typeinfos will use the corresponding serializer to serialize data.
- TypeSerializer serializer = dataType.createSerializer(null);
- ByteArrayOutputStreamWithPos baos = new ByteArrayOutputStreamWithPos();
- DataOutputViewStreamWrapper baosWrapper = new DataOutputViewStreamWrapper(baos);
- serializer.serialize(obj, baosWrapper);
- return pickler.dumps(baos.toByteArray());
- }
- }
- }
-
- private static Object[] primitiveArrayConverter(
- Object array, TypeInformation<?> arrayTypeInfo) {
- Preconditions.checkArgument(arrayTypeInfo instanceof PrimitiveArrayTypeInfo);
- Preconditions.checkArgument(array.getClass().isArray());
- Object[] objects;
- if (PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- boolean[] booleans = (boolean[]) array;
- objects = new Object[booleans.length];
- for (int i = 0; i < booleans.length; i++) {
- objects[i] = booleans[i];
- }
- } else if (PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- byte[] bytes = (byte[]) array;
- objects = new Object[bytes.length];
- for (int i = 0; i < bytes.length; i++) {
- objects[i] = bytes[i];
- }
- } else if (PrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- short[] shorts = (short[]) array;
- objects = new Object[shorts.length];
- for (int i = 0; i < shorts.length; i++) {
- objects[i] = shorts[i];
- }
- } else if (PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- int[] ints = (int[]) array;
- objects = new Object[ints.length];
- for (int i = 0; i < ints.length; i++) {
- objects[i] = ints[i];
- }
- } else if (PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- long[] longs = (long[]) array;
- objects = new Object[longs.length];
- for (int i = 0; i < longs.length; i++) {
- objects[i] = longs[i];
- }
- } else if (PrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- float[] floats = (float[]) array;
- objects = new Object[floats.length];
- for (int i = 0; i < floats.length; i++) {
- objects[i] = floats[i];
- }
- } else if (PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- double[] doubles = (double[]) array;
- objects = new Object[doubles.length];
- for (int i = 0; i < doubles.length; i++) {
- objects[i] = doubles[i];
- }
- } else if (PrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO.equals(arrayTypeInfo)) {
- char[] chars = (char[]) array;
- objects = new Object[chars.length];
- for (int i = 0; i < chars.length; i++) {
- objects[i] = chars[i];
- }
- } else {
- throw new UnsupportedOperationException(
- String.format(
- "Primitive array of %s is not supported in PyFlink yet",
- ((PrimitiveArrayTypeInfo<?>) arrayTypeInfo)
- .getComponentType()
- .getTypeClass()
- .getSimpleName()));
- }
- return objects;
- }
-}
