Re: OOM issues with Python Objects

Posted by Kevin Lam on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/OOM-issues-with-Python-Objects-tp42396p42438.html

Thanks for the response Arvid! Point of clarification, things do NOT OOM when I use the Row subclass. Instead, the code that doesn't use the Row subclass is the code that OOMs (ie. the simple python class). 

On Mon, Mar 22, 2021 at 10:24 AM Arvid Heise <[hidden email]> wrote:

Hi Kevin,

I suspect that this is because Row is not supported as a Python field [1]; it's supposed to be a dict that is mapped to a Row by Flink.

Maybe it runs in some infinite loop while trying serialize and hence the OOM.

Subclassing Row might be an undocumented feature.

I'm also pulling in Dian who knows more about PyFlink.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/datastream-api-users-guide/data_types.html

On Fri, Mar 19, 2021 at 3:55 PM Kevin Lam <[hidden email]> wrote:

Hi all,

I've encountered an interesting issue where I observe an OOM issue in my Flink Application when I use a DataStream of Python Objects, but when I make that Python Object a Subclass of pyflink.common.types.Row and provide TypeInformation, there is no issue.  

For the OOM scenario, no elements get processed, the application runs without printing output and then eventually crashes with java.lang.OutOfMemoryError: Java heap space

Any insights into why this might be happening? Appreciate any help/suggestions.

I've included some code that illustrates the two situations below [0].

Thanks in advance! 

[0]: 

Code Snippet A: OOM scenario

class InputWrapper:

"""Helper class, used to make streams of the same type"""

def __init__(self, key: str, contents: Row = None):

self.key = key

self.contents = contents

x_ds = x_ds.map(

lambda d: InputWrapper(key=d['key'], contents=d))

y_ds = y_ds.map(

lambda o: InputWrapper(key=o['key'], contents=o))

union = x_ds.union(y_ds)

union.print()

Code Snippet B: Working scenario:

class InputWrapper(Row):

"""Helper class, used to make streams of the same type"""

def __init__(self, key: str, contents: Row = None):

super().__init__(key, contents)

x_ds = x_ds.map(

lambda d: InputWrapper(key=d['key'], contents=d),

output_type=InputWrapperTypeInfo())

y_ds = y_ds.map(

lambda o: InputWrapper(key=o['key'], contents=o),

output_type=InputWrapperTypeInfo())

union = x_ds.union(y_ds)

union.print()