[FLINK-14584][python] Support complex data types in Python user-defined functions #10086

HuangXingBo · 2019-11-05T08:45:20Z

What is the purpose of the change

This pr supports ArrayType, MapType, MultisetType, DecimalType In Python UDF

Brief change log

(for example:)

Add DecimalSerializer, BigDecSerializer, BinaryArraySerializer, BinaryMapSerializer
Add DecimalCoder, ArrayCoder, MapCoder, MultisetCoder, CollectionCoder
fix the bug in CharCoder and TinyintCoder

Verifying this change

This change added tests and can be verified as follows:

BigDecSerializerTest, BinaryArraySerializerTest, BinaryMapSerializerTest,DecimalSerializerTest
test_array_coder, test_map_coder, test_multiset_coder, test_decimal_coder in coders_test_common.py
test_all_data_types in test_udf.py

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2019-11-05T08:48:01Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 2746453 (Tue Nov 05 08:48:00 UTC 2019)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-11-05T09:19:56Z

CI report:

2746453 : SUCCESS Build
de45c52 : SUCCESS Build
b837046 : SUCCESS Build
b321d9e : CANCELED Build
43f2a21 : FAILURE Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

WeiZhong94

@HuangXingBo Thanks for your PR! It looks good overall, here is a few comments.

WeiZhong94 · 2019-12-03T07:23:53Z

flink-python/pyflink/fn_execution/coder_impl.py

+    def decode_from_stream(self, in_stream, nested):
+        size = in_stream.read_bigendian_int32()
+        elements = [self._elem_coder.decode_from_stream(in_stream, nested)
+                    if not not in_stream.read_byte() else None for _ in range(size)]


remove "not not"?

WeiZhong94 · 2019-12-03T07:31:20Z

flink-python/pyflink/fn_execution/coder_impl.py

+        for _ in range(size):
+            key = self._key_coder.decode_from_stream(in_stream, nested)
+            is_null = not not in_stream.read_byte()
+            if is_null:


use in_stream.read_byte() directly or bool(in_stream.read_byte())?

WeiZhong94 · 2019-12-03T07:47:27Z

flink-python/pyflink/fn_execution/coder_impl.py

+        return map_value
+
+    def __repr__(self):
+        return 'MapCoderImpl[%s]' % ' : '.join([str(self._key_coder), str(self._value_coder)])


use repr() instead of str()?

WeiZhong94 · 2019-12-03T07:53:24Z

flink-python/pyflink/fn_execution/coder_impl.py

+
+    def encode_to_stream(self, value, out_stream, nested):
+        dict_value = self.multiset_to_dict(value)
+        out_stream.write_bigendian_int32(len(dict_value))


This part is duplicated with MapCoderImpl, can we reuse it?

WeiZhong94 · 2019-12-03T08:51:25Z

flink-python/pyflink/fn_execution/coder_impl.py

+class DecimalCoderImpl(StreamCoderImpl):
+
+    def __init__(self, precision, scale):
+        decimal.getcontext().prec = precision


Maybe we should hold a individual context object here and replace current context at the beginning of encode/decode and restore users' context at the end of encode/decode?

WeiZhong94 · 2019-12-03T10:15:22Z

flink-python/pyflink/fn_execution/coders.py

+        self._elem_coder = elem_coder
+        super(ArrayCoder, self).__init__(elem_coder)
+
+    def _impl_coder(self):


How about override _create_impl directly?

WeiZhong94 · 2019-12-03T10:23:41Z

flink-python/pyflink/fn_execution/tests/coders_test_common.py

+
+    def test_decimal_coder(self):
+        from decimal import Decimal
+        coder = DecimalCoder()


How about test with different precision?

WeiZhong94 · 2019-12-04T07:37:12Z

.../java/org/apache/flink/table/runtime/typeutils/serializers/python/BinaryArraySerializer.java

+ * Currently Python doesn't support BinaryArray natively, so we can't use BaseArraySerializer in blink directly.
+ */
+@Internal
+public class BinaryArraySerializer<K> extends BaseArraySerializer {


If we extends BaseArraySerializer, it seems the type parameter "K" is unnecessary?

The name "BinaryArraySerializer" is not accurate, maybe "PythonBaseArraySerializer" is better?

Make sense.What about naming it BaseArraySerializer?

WeiZhong94 · 2019-12-04T07:56:59Z

...in/java/org/apache/flink/table/runtime/typeutils/serializers/python/BinaryMapSerializer.java

+ * Currently Python doesn't support BinaryMap natively, so we can't use BaseArraySerializer in blink directly.
+ */
+@Internal
+public class BinaryMapSerializer<K, V> extends BaseMapSerializer {


HuangXingBo · 2019-12-04T12:13:09Z

Thanks a lot for @WeiZhong94 review. I have addressed comments at the latest commit.

hequn8128

@HuangXingBo Thanks a lot for the PR. Some quick feedback below. Will leave more later.

hequn8128 · 2019-12-05T07:18:20Z

flink-python/pyflink/table/tests/test_udf.py

+                'multiset_param is wrong value %s !' % multiset_param
+            return multiset_param
+
+        def create_multiset_func():


It's not good to use udf to create input data. Because, the data will be given to multiset_func directly within Python, i.e., the serialize method of the java serializer can not be tested.

hequn8128 · 2019-12-05T07:39:54Z

flink-python/pyflink/table/tests/test_udf.py

            .insert_into("Results")
        self.t_env.execute("test")
        actual = source_sink_utils.results()
        self.assert_equals(actual,
                           ["1,null,1,true,32767,-2147483648,1.23,1.98932,"
-                            "[102, 108, 105, 110, 107],pyflink,2014-09-13"])
+                            "[102, 108, 105, 110, 107],pyflink,2014-09-13,"
+                            "[1, 2, 3],{1=flink, 2=pyflink},{1=2, 2=1},"


For multiset, we output a map? What's the behavior of java/scala?
I think maybe we can support multiset in python later if we find there is a need. For python, there are no python types corresponding to multiset. Furthermore, a user can somehow use a map type to achieve multiset type.

Yes.Neither array, set, or map can represent the multiset very conveniently in Python.We can support support multiset in python later if we find a good structure to express multiset.

hequn8128 · 2019-12-05T09:09:57Z

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

@@ -70,6 +89,43 @@ public static TypeSerializer toBlinkTypeSerializer(LogicalType logicalType) {
 		return logicalType.accept(new LogicalTypeToProtoTypeConverter());
 	}

+	/**
+	 * Convert LogicalType to conversion class.


This class is only used for flink planner. Maybe change the comment to: "Convert LogicalType to conversion class for flink planner"?

hequn8128 · 2019-12-05T09:43:38Z

flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/PythonTypeUtils.java

+		public TypeSerializer visit(ArrayType arrayType) {
+			LogicalType elementType = arrayType.getElementType();
+			TypeSerializer<?> elementTypeSerializer = elementType.accept(this);
+			Class<?> elementClass = LogicalTypeToConversionClassConverter.INSTANCE.visit(elementType);


It seems we don't need to add the LogicalTypeToConversionClassConverter class. We can use TypeConversions to convert the LogicalType to the array TypeInformation and then convert to the serializer.

As we disscussed offline, logicalType does not contain class information, so we can't use LogicalTypeToConversionClassConverter to get the correct conversion class information of DateType, TimeType,TimestampType and ArrayType

hequn8128 · 2019-12-05T09:59:51Z

flink-python/pyflink/fn_execution/coder_impl.py

+    def __init__(self, elem_coder):
+        self._elem_coder = elem_coder
+
+    def encode_to_stream(self, value, out_stream, nested):


It is more efficient to use null mask to handle null values rather than adding a boolean value for every element. However, it is the behavior of Java.

hequn8128 · 2019-12-05T10:01:07Z

flink-python/pyflink/table/tests/test_udf.py

@@ -319,6 +319,30 @@ def date_func(date_param):
                'date_param is wrong value %s !' % date_param
            return date_param

+        def array_func(array_param):
+            assert array_param == [1, 2, 3], \


Can we test nested array?

hequn8128 · 2019-12-05T10:10:27Z

...in/java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseArraySerializer.java

+	}
+
+	@Override
+	public BaseArray deserialize(BaseArray reuse, DataInputView source) throws IOException {


Can we reuse the BaseArray here?

hequn8128 · 2019-12-05T10:11:46Z

...in/java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseArraySerializer.java

+	/**
+	 * {@link TypeSerializerSnapshot} for {@link BaseArraySerializer}.
+	 */
+	public static final class BaseArraySerializerSnapshot implements TypeSerializerSnapshot<BaseArray> {


We can reuse the BaseArraySerializerSnapshot in the base class.

The method testSnapshotConfigurationAndReconfigure of SerializerTestBase will test the class of serializer, so we can't reuse BaseArraySerializerSnapshot directly.

hequn8128

@HuangXingBo Some more comments.

hequn8128 · 2019-12-06T02:41:10Z

...main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseMapSerializer.java

+
+	@Override
+	public void serialize(BaseMap map, DataOutputView target) throws IOException {
+		BinaryMap binaryMap = (BinaryMap) map;


The BaseMap map can be a GenericMap. Take a look at the public BaseMap copy(BaseMap from) method in the Base class(org.apache.flink.table.runtime.typeutils.BaseMapSerializer)

hequn8128 · 2019-12-06T02:45:54Z

.../java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseMapSerializerTest.java

+	}
+
+	@Override
+	protected BaseMap[] getTestData() {


Also test GenericMap?

hequn8128 · 2019-12-06T02:58:35Z

...in/java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseArraySerializer.java

+	 * {@link TypeSerializerSnapshot} for {@link BaseArraySerializer}.
+	 */
+	public static final class BaseArraySerializerSnapshot implements TypeSerializerSnapshot<BaseArray> {
+		private static final int CURRENT_VERSION = 3;


Current version to 1?

hequn8128 · 2019-12-06T02:58:46Z

...main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BaseMapSerializer.java

+	 * {@link TypeSerializerSnapshot} for {@link BaseMapSerializer}.
+	 */
+	public static final class BaseMapSerializerSnapshot implements TypeSerializerSnapshot<BaseMap> {
+		private static final int CURRENT_VERSION = 3;


hequn8128 · 2019-12-06T03:22:38Z

...main/java/org/apache/flink/table/runtime/typeutils/serializers/python/DecimalSerializer.java

+ * for performance reasons in Python deserialization.
+ */
+@Internal
+public class DecimalSerializer extends TypeSerializer<Decimal> {


Extends from the blink DecimalSerializer?

DecimalSerializer is final class, so we can't extends from the blink DecimalSerializer.

hequn8128 · 2019-12-06T03:23:09Z

.../main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BigDecSerializer.java

+ * performance reasons in Python deserialization.
+ */
+@Internal
+public class BigDecSerializer extends TypeSerializerSingleton<BigDecimal> {


Extends from the BigDecSerializer in flink-core?

hequn8128 · 2019-12-06T03:48:59Z

flink-python/pyflink/table/tests/test_udf.py

        t = self.t_env.from_elements(
            [(1, None, 1, True, 32767, -2147483648, 1.23, 1.98932,
-              bytearray(b'flink'), 'pyflink', datetime.date(2014, 9, 13))],
+              bytearray(b'flink'), 'pyflink', datetime.date(2014, 9, 13),
+              [1, 2, 3], {1: 'flink', 2: 'pyflink'}, decimal.Decimal('1000000000000000000.05'))],


Test decimal with (38,18)?

HuangXingBo · 2019-12-06T10:58:09Z

Thanks a lot for @hequn8128 review, I have addressed the comments at latest commit.

hequn8128 · 2019-12-06T13:33:24Z

@HuangXingBo Thanks a lot for the update. Will merge this once travis passed.

This closes apache#10086.

rmetzger added the review=description? label Nov 5, 2019

rmetzger added the component=API/Python label Nov 5, 2019

HuangXingBo force-pushed the FLINK-14584 branch from 2746453 to de45c52 Compare November 28, 2019 10:05

WeiZhong94 reviewed Dec 4, 2019

View reviewed changes

HuangXingBo force-pushed the FLINK-14584 branch from de45c52 to b837046 Compare December 4, 2019 12:11

hequn8128 reviewed Dec 5, 2019

View reviewed changes

hequn8128 reviewed Dec 6, 2019

View reviewed changes

HuangXingBo added 9 commits December 6, 2019 15:48

[FLINK-14584][python] Support ArrayType

767eb8a

[FLINK-14584][python]-1-fix Support ArrayType

22d56de

[FLINK-14584][python]-2-fix Support ArrayType

14589c0

[FLINK-14584][python] Support MapType

199f45f

[FLINK-14584][python]-1-fix Support MapType

baa4731

[FLINK-14584][python]-2-fix Support MapType

f2f4588

[FLINK-14584][python] Support DecimalType

4df1ba1

[FLINK-14584][python]-1-fix Support DecimalType

c4d197e

[FLINK-14584][python]-2-fix Support DecimalType

a4886a3

HuangXingBo force-pushed the FLINK-14584 branch from b837046 to b321d9e Compare December 6, 2019 10:02

HuangXingBo added 2 commits December 6, 2019 20:08

[FLINK-14584][python]-3-fix Support DecimalType

ccee0c1

[FLINK-14584][python] Fix impletation bug of TinyIntCoder and CharCoder

43f2a21

HuangXingBo force-pushed the FLINK-14584 branch from b321d9e to 43f2a21 Compare December 6, 2019 12:09

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Dec 6, 2019

[FLINK-14584][python] Fix impletation bug of TinyIntCoder and CharCoder

2e7e06c

This closes apache#10086.

hequn8128 closed this in f208354 Dec 7, 2019

[FLINK-14584][python] Support complex data types in Python user-defined functions #10086

[FLINK-14584][python] Support complex data types in Python user-defined functions #10086

Conversation

HuangXingBo commented Nov 5, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Nov 5, 2019

Automated Checks

Review Progress

flinkbot commented Nov 5, 2019 • edited

CI report:

WeiZhong94 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Dec 4, 2019

hequn8128 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hequn8128 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HuangXingBo commented Dec 6, 2019

hequn8128 commented Dec 6, 2019

flinkbot commented Nov 5, 2019 •

edited