1. 全景图
2. 合并
根据attribute合并flowfile:
合并 json, 并增加code，message等：
3. 计算方差：
在ExecuteScript里只能用纯python, 很多第三方包都不能用；并把计算的值插入到json里，输出。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
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：
可视化图：
NIFI 中国社区 QQ群：595034369
本文转自疯吻IT博客园博客，原文链接：http://www.cnblogs.com/fengwenit/p/5823827.html，如需转载请自行联系原作者
