[
https://issues.apache.org/jira/browse/FLINK-30168?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu closed FLINK-30168.
---------------------------
Fix Version/s: 1.17.0
1.16.1
1.15.4
Assignee: Dian Fu (was: Xingbo Huang)
Resolution: Fixed
Fixed in:
- master via 46757739cf50c1e7b7305a4bc9cf779bb1945a1f
- release-1.16 via 46c91ed4bc22e2de3a662d52b11ade8ed64dba0b
- release-1.15 via cdecc21cad9f78b1555a0e2f5d7f1398949e7193
> PyFlink Deserialization Error with Object Array
> -----------------------------------------------
>
> Key: FLINK-30168
> URL: https://issues.apache.org/jira/browse/FLINK-30168
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 1.16.0, 1.15.2
> Reporter: Yunfeng Zhou
> Assignee: Dian Fu
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.17.0, 1.16.1, 1.15.4
>
>
> When it is attempted to collect object array records from a DataStream in
> PyFlink, an exception like follows would be thrown
> {code:java}
> data = 0, field_type = DenseVectorTypeInfo
> def pickled_bytes_to_python_converter(data, field_type):if
> isinstance(field_type, RowTypeInfo):
> row_kind = RowKind(int.from_bytes(data[0], 'little'))
> data = zip(list(data[1:]), field_type.get_field_types())
> fields = []for d, d_type in data:
> fields.append(pickled_bytes_to_python_converter(d, d_type))
> row = Row.of_kind(row_kind, *fields)return rowelse:
> > data = pickle.loads(data)
> E TypeError: a bytes-like object is required, not 'int'{code}
> I found that this error is invoked because PyFlink deals with object arrays
> differently on Java side and Python side.
>
> On Java side
> (org.apache.flink.api.common.python.PythonBridgeUtils.getPickledBytesFromJavaObject)
> {code:java}
> ...
> else if (dataType instanceof BasicArrayTypeInfo || dataType instanceof
> PrimitiveArrayTypeInfo) {
> # recursively deal with array elements
> } ...
> else {
> # ObjectArrayTypeInfo is here
> TypeSerializer serializer = dataType.createSerializer(null);
> ByteArrayOutputStreamWithPos baos = new ByteArrayOutputStreamWithPos();
> DataOutputViewStreamWrapper baosWrapper = new
> DataOutputViewStreamWrapper(baos); serializer.serialize(obj, baosWrapper);
> return pickler.dumps(baos.toByteArray());
> }
> {code}
>
> On python side(pyflink.datastream.utils.pickled_bytes_to_python_converter)
> {code:java}
> ...
> elif isinstance(field_type,
> (BasicArrayTypeInfo, PrimitiveArrayTypeInfo, ObjectArrayTypeInfo)):
> element_type = field_type._element_type
> elements = []
> for element_bytes in data:
> elements.append(pickled_bytes_to_python_converter(element_bytes,
> element_type))
> return elements{code}
>
> Thus a possible fix for this bug is to align PyFlink's behavior on Java side
> and Python side.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
