python之如何将参数传递给Nifi中的python脚本

也许这是一个愚蠢的问题，但我不得不问。

我在 Nifi 中有一个 Collect_data 处理器，它将消息流式传输到另一个使用 python 脚本解析并创建 json 文件的进程。问题是我不知道 python 脚本中函数的输入是什么。如何将这些消息(16 位数字)从 Collect_data 处理器传递到下一个处理器包含 python 脚本。有什么好的、基本的例子吗？

我已经在网上寻找一些示例，但并没有真正理解。

import datetime 
import hashlib 
from urlparse import urlparse, parse_qs 
import sys 
from urlparse import urlparse, parse_qs 
from datetime import * 
import json 
import java.io 
from org.apache.commons.io import IOUtils 
from java.nio.charset import StandardCharsets 
from org.apache.nifi.processor.io import StreamCallback 
from time import time 
 
 
def parse_zap(inputStream, outputStream): 
    data = inputStream 
    buf = (hashlib.sha256(bytearray.fromhex(data)).hexdigest()) 
    buf = int(buf, 16) 
    buf_check = str(buf) 
    if buf_check[17] == 2: 
        pass 
    datetime_now = datetime.now() 
    log_date = datetime_now.isoformat() 
    try: 
        mac = buf_check[7:14].upper() 
        ams_id = buf_check[8:] 
        action = buf_check[3:4] 
        time_a = int(time()) 
        dict_test = { 
        "user": { 
            "guruq" : 'false' 
        }, 
        "device" : { 
            "type" : "siolbox", 
            "mac": mac 
        }, 
        "event" : { 
            "origin" : "iptv", 
            "timestamp": time_a, 
            "type": "zap", 
            "product-type" : "tv-channel", 
            "channel": { 
                "id" : 'channel_id', 
                "ams-id": ams_id 
            }, 
            "content": { 
                "action": action 
            } 
        } 
        } 
        return dict_test 
    except Exception as e: 
        print('%s nod PARSE 500 \"%s\"' % (log_date, e)) 

谢谢，我没看错，但现在我无法创建输出。 提前致谢。

请您参考如下方法:

看看这个脚本:

import json 
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): 
    text = IOUtils.readLines(inputStream, StandardCharsets.UTF_8) 
    for line in text[1:]: 
        outputStream.write(line + "\n")  
 
flowFile = session.get() 
if (flowFile != None): 
  flowFile = session.write(flowFile,PyStreamCallback()) 
  flowFile = session.putAttribute(flowFile, "filename", flowFile.getAttribute('filename').split('.')[0]+'_translated.json') 
  session.transfer(flowFile, REL_SUCCESS) 

它从一个属性中获取要从流文件中删除的行数，然后获取流文件并在没有这些行的情况下再次写入它，这很简单，也是两者的一个很好的例子，如何使用这些属性，以及如何使用流程文件。

根据您更新后的代码，您的代码必须如下所示:

import datetime 
import hashlib 
from urlparse import urlparse, parse_qs 
import sys 
from urlparse import urlparse, parse_qs 
from datetime import * 
import json 
import java.io 
from org.apache.commons.io import IOUtils 
from java.nio.charset import StandardCharsets 
from org.apache.nifi.processor.io import StreamCallback 
from time import time 
 
 
class PyStreamCallback(StreamCallback): 
  def __init__(self): 
        pass 
  def process(self, inputStream, outputStream): 
    data = inputStream 
    buf = (hashlib.sha256(bytearray.fromhex(data)).hexdigest()) 
    buf = int(buf, 16) 
    buf_check = str(buf) 
    if buf_check[17] == 2: 
        pass 
    datetime_now = datetime.now() 
    log_date = datetime_now.isoformat() 
    try: 
        mac = buf_check[7:14].upper() 
        ams_id = buf_check[8:] 
        action = buf_check[3:4] 
        time_a = int(time()) 
        dict_test = { 
        "user": { 
            "guruq" : 'false' 
        }, 
        "device" : { 
            "type" : "siolbox", 
            "mac": mac 
        }, 
        "event" : { 
            "origin" : "iptv", 
            "timestamp": time_a, 
            "type": "zap", 
            "product-type" : "tv-channel", 
            "channel": { 
                "id" : 'channel_id', 
                "ams-id": ams_id 
            }, 
            "content": { 
                "action": action 
            } 
        } 
        } 
        return dict_test 
    except Exception as e: 
        print('%s nod PARSE 500 \"%s\"' % (log_date, e)) 
 
flowFile = session.get() 
if (flowFile != None): 
  flowFile = session.write(flowFile,PyStreamCallback()) 
  flowFile = session.putAttribute(flowFile, "filename", flowFile.getAttribute('filename').split('.')[0]+'_translated.json') 
  session.transfer(flowFile, REL_SUCCESS)