I'm working on a custom Python script to execute inside 'ExecuteScript' processor in NIFI.   This processor receives input JSON flowfiles and only those that are present in variable filter inside script, should pass through. So the idea is filter JSON based on one column.   My current script looks like this:     import json flowFile = session.get() if flowFile is not None: filter = ['1000007', '1000008'] json_content = json.loads(flowFile.read().decode('utf-8')) if json_content['Customer_AccountID'] in patata: filtered_content = json.dumps(json_content) flowFile.write(bytearray(filtered_content, 'utf-8')) session.transfer(flowFile, REL_SUCCESS) else: session.remove(flowFile) session.transfer(flowFile, REL_FAILURE)     For now is not filtering anything, no output at all. Any idea why?   ... View more

I'm trying to using NIFI dynamically filter flow Files in JSON format and keep only those values that are present is one SQL table.   I'm using QueryRecord processor, with query like: SELECT * FROM FLOWFILE WHERE ${key} = ${value} only those values that are present in QueryDatabaseTable variables should be kept.   Flow Files to be filtered looks like these:   [ { "CustomerID" : 1, "OrderNbr" : "12", "Amount":100 } ] CustomerID field is the one to check and filter data.   My current NIFI flow:      ... View more

I'm looking for breaking following nested JSON file and transform it into a SQL prepared format.   Input JSON file:       { "Product1": { "Purchase": 31 }, "Product2": { "Purchase": 6213, "Cancel": 1988, "Change": 3702, "Renewal": 5934 } }       Desired output:        [ { "product": "Product1", "Purchase": 31 }, { "product": "Product2", "Purchase": 6213, "Cancel": 1988, "Change": 3702, "Renewal": 5934 } ]     ... View more

I'm looking for breaking nested JSON file and try to flat it to fit into a SQL database.   Current Json:   { "content": { "failedPerProductLineAndReason": { "Product1": { "Downsizing licenses is not allowed": 1 } } } }    Expected outcome:  { "ErrorType": "failedPerProductLineAndReason", "product": "Product1", "error": "Downsizing licenses is not allowed", "quantity": 1 }   ... View more

I wanted to wait until 'ExecuteSQL' processor finishes inserting big file into SQL database (table: bu_service_template), then move to other section. 'ExecuteSQL' processor inserts row by row content of a file. I wanted to wait until file is fully inserted and then move to a next file. My current flow: wait flow ... View more

How can I break and flatten nested JSON with arrays using Jolt transformations from:       { "product": [ "data - 1", "data - 2" ], "Purchase": [ 2, 2 ], "Renewal": 1 }         to         [ { "product": "data -1", "Purchase": 1, "Renewal": 1 }, { "product": "data -2", "Purchase": 2, } ]         My current jolt transformation doesn't give me fully accurate transformation         [ { "operation": "shift", "spec": { "*": { "*": { "@": "[#2].&2" } } } } ]         ... View more