You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2022/11/23 12:17:00 UTC
[jira] [Updated] (FLINK-30124) GenericType is not supported in PyFlink currently
[ https://issues.apache.org/jira/browse/FLINK-30124?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
ASF GitHub Bot updated FLINK-30124:
-----------------------------------
Labels: pull-request-available (was: )
> GenericType<java.util.Map> is not supported in PyFlink currently
> ----------------------------------------------------------------
>
> Key: FLINK-30124
> URL: https://issues.apache.org/jira/browse/FLINK-30124
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 1.15.1
> Reporter: Yunfeng Zhou
> Priority: Major
> Labels: pull-request-available
>
> When we add and execute the following test case to flink-ml-python/pyflink/ml/lib/classification/tests/test_naivebayes.py of the Flink ML repository,
> {code:java}
> def test_get_model_data(self):
> model_data = self.estimator.fit(self.train_data).get_model_data()[0]
> self.t_env.to_data_stream(model_data).execute_and_collect().next(){code}
> The following exception would be thrown.
>
> {code:java}
> j_type_info = JavaObject id=o698
> def _from_java_type(j_type_info: JavaObject) -> TypeInformation:
> gateway = get_gateway()
> JBasicTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo
>
> if _is_instance_of(j_type_info, JBasicTypeInfo.STRING_TYPE_INFO):
> return Types.STRING()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BOOLEAN_TYPE_INFO):
> return Types.BOOLEAN()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BYTE_TYPE_INFO):
> return Types.BYTE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.SHORT_TYPE_INFO):
> return Types.SHORT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INT_TYPE_INFO):
> return Types.INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.LONG_TYPE_INFO):
> return Types.LONG()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.FLOAT_TYPE_INFO):
> return Types.FLOAT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.DOUBLE_TYPE_INFO):
> return Types.DOUBLE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.CHAR_TYPE_INFO):
> return Types.CHAR()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_INT_TYPE_INFO):
> return Types.BIG_INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_DEC_TYPE_INFO):
> return Types.BIG_DEC()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INSTANT_TYPE_INFO):
> return Types.INSTANT()
>
> JSqlTimeTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo
> if _is_instance_of(j_type_info, JSqlTimeTypeInfo.DATE):
> return Types.SQL_DATE()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIME):
> return Types.SQL_TIME()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIMESTAMP):
> return Types.SQL_TIMESTAMP()
>
> JPrimitiveArrayTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo \
> .PrimitiveArrayTypeInfo
>
> if _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BYTE())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.SHORT())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.INT())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.LONG())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.FLOAT())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.DOUBLE())
> elif _is_instance_of(j_type_info, JPrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.CHAR())
>
> JBasicArrayTypeInfo = gateway.jvm.org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo
>
> if _is_instance_of(j_type_info, JBasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.BYTE())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.SHORT_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.SHORT())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.INT_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.INT())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.LONG_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.LONG())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.FLOAT_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.FLOAT())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.DOUBLE())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.CHAR_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.CHAR())
> elif _is_instance_of(j_type_info, JBasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.STRING())
>
> JObjectArrayTypeInfo = gateway.jvm.org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo
> if _is_instance_of(j_type_info, JObjectArrayTypeInfo):
> return Types.OBJECT_ARRAY(_from_java_type(j_type_info.getComponentInfo()))
>
> JPickledBytesTypeInfo = gateway.jvm \
> .org.apache.flink.streaming.api.typeinfo.python.PickledByteArrayTypeInfo\
> .PICKLED_BYTE_ARRAY_TYPE_INFO
> if _is_instance_of(j_type_info, JPickledBytesTypeInfo):
> return Types.PICKLED_BYTE_ARRAY()
>
> JRowTypeInfo = gateway.jvm.org.apache.flink.api.java.typeutils.RowTypeInfo
> if _is_instance_of(j_type_info, JRowTypeInfo):
> j_row_field_names = j_type_info.getFieldNames()
> j_row_field_types = j_type_info.getFieldTypes()
> row_field_types = [_from_java_type(j_row_field_type) for j_row_field_type in
> j_row_field_types]
> row_field_names = [field_name for field_name in j_row_field_names]
> return Types.ROW_NAMED(row_field_names, row_field_types)
>
> JTupleTypeInfo = gateway.jvm.org.apache.flink.api.java.typeutils.TupleTypeInfo
> if _is_instance_of(j_type_info, JTupleTypeInfo):
> j_field_types = []
> for i in range(j_type_info.getArity()):
> j_field_types.append(j_type_info.getTypeAt(i))
> field_types = [_from_java_type(j_field_type) for j_field_type in j_field_types]
> return TupleTypeInfo(field_types)
>
> JMapTypeInfo = get_gateway().jvm.org.apache.flink.api.java.typeutils.MapTypeInfo
> if _is_instance_of(j_type_info, JMapTypeInfo):
> j_key_type_info = j_type_info.getKeyTypeInfo()
> j_value_type_info = j_type_info.getValueTypeInfo()
> return MapTypeInfo(_from_java_type(j_key_type_info), _from_java_type(j_value_type_info))
>
> JListTypeInfo = get_gateway().jvm.org.apache.flink.api.java.typeutils.ListTypeInfo
> if _is_instance_of(j_type_info, JListTypeInfo):
> j_element_type_info = j_type_info.getElementTypeInfo()
> return ListTypeInfo(_from_java_type(j_element_type_info))
>
> JExternalTypeInfo = gateway.jvm.org.apache.flink.table.runtime.typeutils.ExternalTypeInfo
> if _is_instance_of(j_type_info, JExternalTypeInfo):
> TypeInfoDataTypeConverter = \
> gateway.jvm.org.apache.flink.table.types.utils.LegacyTypeInfoDataTypeConverter
> return ExternalTypeInfo(_from_java_type(
> TypeInfoDataTypeConverter.toLegacyTypeInfo(j_type_info.getDataType())))
>
> > raise TypeError("The java type info: %s is not supported in PyFlink currently." % j_type_info)
> E TypeError: The java type info: GenericType<java.util.Map> is not supported in PyFlink currently.
> {code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)