flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Chesnay Schepler (JIRA)" <j...@apache.org>
Subject [jira] [Created] (FLINK-2501) [py] Remove the need to specify types for transformations
Date Sat, 08 Aug 2015 07:40:45 GMT
Chesnay Schepler created FLINK-2501:

             Summary: [py] Remove the need to specify types for transformations
                 Key: FLINK-2501
                 URL: https://issues.apache.org/jira/browse/FLINK-2501
             Project: Flink
          Issue Type: Improvement
          Components: Python API
            Reporter: Chesnay Schepler

Currently, users of the Python API have to provide type arguments when using a UDF, like so:

d1.map(Mapper(), (INT, STRING))

Instead, it would be really convenient to be able to do this:


The intention behind this issue is convenience, and it's also not really pythonic to specify

Before I'll go into possible solutions, let me summarize the way these type arguments are
currently used, and in general how types are handled:

The type argument passed is actually an object of the type it represents, as INT is a constant
int value, whereas STRING is a constant string value. You could as well write the following
and it would still work.
d1.map(Mapper(), (1, "ImNotATypInfo"))
This object is transmitted to the java side during the plan binding (and is now an actual
Tuple2<Integer, String>), then passed to the type extractor, and the resulting TypeInformation
saved in the java counterpart of the udf, which all implement the ResultTypeQueryable interface.

The TypeInformation object is only used by the Java API, python never touches it. Instead,
at runtime, the serializers used between python and java check the classes of the values passed
and are thus generated dynamically.
This means that, if a UDF does not pass the type it claims to pass, the Python API wont complain,
but the underlying java API will when it's serializers fail.

Now let's talk solutions.

In discussions on the mailing list, pretty much 2 proposals were made:
# Add a way to disable/circumvent type checks during the plan phase in the Java API and generate
serializers dynamically.
# Have objects always in serialized form on the java side, stored in a single bytearray or
Tuple2 containing a key/value pair.

These proposals vary wildly in the changes necessary to the system:
# "How can we change the Java API to support this?"
This proposal would hardly change the way the Python API works, or even touch the related
source code. It mostly deals with the Java API. Since I'm not to familiar with the Plan processing
life-cycle on the java side I can't assess which classes would have to be changed.
# "How can we make this work within the limits of the Java API?"
is the exact opposite, it changes nothing in the Java API. Instead, the following issues would
have to be solved:
* Alter the plan to extract keys before keyed operations, while hiding these keys from the
UDF. This is exactly how KeySelectors (will) work, and as such is generally solved. In fact,
this solution would make a few things easier in regards to KeySelectors.
* Rework all operations that currently rely on Java API functions, that need deserialized
data, for example Projections or the upcoming Aggregations; 
This generally means implementing them in python, or with special java UDF's (they could de-/serialize
data within the udf call, or work on serialized data).
* Change (De)Serializers accordingly
* implement a reliable, not all-memory-consuming sorting mechanism on the python side

Personally i prefer the second option, as it
# does not modify the Java API, it works within it's well-tested limits
# Plan changes are similar to issues that are already worked on (KeySelectors)
# Sorting implementation was necessary anyway (for chained reducers)
# having data in serialized form was a performance-related consideration already

While the first option could work, and most likely require less work, i feel like many of
the things required for option 2 will be implemented eventually anyway.

This message was sent by Atlassian JIRA

View raw message