用Nifi合并二个API、计算并生成新的API

import simplejson as json #from scipy.stats import f_oneway import java.io 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): pass def process(self, inputStream, outputStream): jsonData = IOUtils.toString(inputStream, StandardCharsets.UTF_8) data = json.loads(jsonData) values = [float(i[‘fltValue‘]) for i in data["data"]] firsts = [float(i[‘first‘]) for i in data["data"]] seconds = [float(i[‘second‘]) for i in data["data"]] def stdDeviation(a): count = len(a) if count < 2: return 0 avg = sum(a)/count result = 0.0 for i in a: result += (i - avg)**2 return (result/(count - 1))**0.5 v = stdDeviation(values) f = stdDeviation(firsts) s = stdDeviation(seconds) data["valueDev"] = v data["firstDev"] = f data["secondDev"] = s outputStream.write(bytearray(json.dumps(data, indent=4).encode(‘utf-8‘))) flowFile = session.get() if (flowFile != None): flowFile = session.write(flowFile,PyStreamCallback()) session.transfer(flowFile, REL_SUCCESS)

?

4. 最终效果：

第一个API：

第二个API：

?

最后合并生成的API：

可视化图：

?