CODEWITHSUNDEEP
heap-memoryapache-nifi
Cong
Cong

Reputation: 479

Nifi ExecuteScript with python: OOM exception

One Trigger Processor->GenerateTableFetch->ExecuteSQLRecord->UpdateAttribute->ExecuteScript(python)->.... as pic enter image description here

But Getting an OutOfMemory exception in ExecuteScript,What can i do?TIA

Log

2019-11-15 16:37:33,466 ERROR [pool-14-thread-1] o.a.n.c.r.WriteAheadFlowFileRepository Unable to checkpoint FlowFile Repository due to java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space

Script

import json
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback
class PyStreamCallback(StreamCallback):
    def __init__(self, flowfile):
        self.ff = flowfile
        pass

    def process(self, inputStream, outputStream):
        # prop_col_names = self.ff.getAttribute('prop_col_names')

        prop_col_names = ['FACTORYNO']
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        obj = json.loads(text)

        results_prop = []
        results_basic = []

        for row in obj:
            main_id= row['COMBINEPICKUPNO']
            # main_prop_table
            for col_name in prop_col_names:
                # TODO: read from variable
                d = {}
                if col_name == 'FACTORYNO':
                    d = {'VALUE': row[col_name],
                         'ID': col_name,
                         }
                results_prop.append(d)
            # TODO: read from variable
            basic_info = {
                'ID': main_id,
            }
            results_basic.append(basic_info)

            results_pxp.append(pxp_info)
        data = {
            'prop': results_prop,
            'basic': results_basic
        }
        outputStream.write(bytearray(json.dumps(data, separators=(',', ':')).encode('utf-8')))

flowFile = session.get()
if (flowFile != None):
    flowFile = session.write(flowFile, PyStreamCallback(flowFile))
    session.transfer(flowFile, REL_SUCCESS)

Java 1.8 Nifi 1.9.2

Upvotes: 3

Views: 654

Answers (1)

mattyb
mattyb

Reputation: 12083

If you're reading huge files into memory, daggett is right that you can run out of memory.

Alternatively, I believe there are known memory leak(s) in Jython, but I couldn't find a page for the one I was looking for. You can try to set the system property python.options.internalTablesImpl to weak in bootstrap.conf and restarting NiFi.

As another possible option, since your script isn't too long, you could port the code to Groovy, it has many similar idioms as Python so hopefully wouldn't be too complicated.

Upvotes: 3

Related Questions