nifiexecuteScript 中的 javascript 错误

Question

我在获取executeScript nifi 进程的javascript 脚本时遇到问题，希望得到这方面的帮助。目标是传递一个包含 json 对象的流文件。我需要在不事先知道内容/字段的情况下解析此 json，并将其传递到将其传递到下一个进程(即 MergeContent)的流文件中，并计算流文件的数量。 尝试测试脚本并得到以下错误:

 nifi.script.ExecuteScript - ExecuteScript[id=bd6842e9-e3a4-4d88-a59d- 
 7da1d74d109b] ExecuteScript[id=bd6842e9-e3a4-4d88-a59d-7da1d74d109b] 
 failed to process due to 
 org.apache.nifi.processor.exception.ProcessException: 
 javax.script.ScriptException: <eval>:21:17 Expected : but found value
         let value = json[key];
             ^ in <eval> at line number 21 at column number 17; rolling 
 back session: org.apache.nifi.processor.exception.ProcessException: 
 javax.script.ScriptException: <eval>:21:17 Expected : but found value

我对 javascript 不太熟悉，因此非常感谢您的帮助。

flowFile = session.get();
if (flowFile != null) {

var StreamCallback = 
Java.type("org.apache.nifi.processor.io.StreamCallback");
var IOUtils = Java.type("org.apache.commons.io.IOUtils");
var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
var transformed_message = {};
var error = false;
var line = "ops_testQueue";

flowFile = session.write(flowFile, new StreamCallback(function 
  (inputStream, outputStream) {

    var content = IOUtils.toString(inputStream, 
  StandardCharsets.UTF_8); // message or content
    var message_content = {};
    try {
        message_content = JSON.parse(content);
        if(Array.isArray(message_content)){

    }
    Object.keys(message_content).forEach((key) => {
         var value = json[key];
         result.push(key + '=' + value);
         var jkey = "," +  "\"" + key + "\""  + '=' + value   
        });
    line = line + jkey +
            " value=" + "1"
            + " " + Date.now() * 1000000;


        // Write output content
        if (transformed_message) {
            outputStream.write(line.getBytes(StandardCharsets.UTF_8));
        }
    } catch (e) {
        error = true;
        outputStream.write(content.getBytes(StandardCharsets.UTF_8));
    }
 }));
 if (transformed_message.post_state) {
    flowFile = session.putAttribute(flowFile, "type", 
    transformed_message.type);
}

if (error) {
    session.transfer(flowFile, REL_FAILURE)
} else {
    session.transfer(flowFile, REL_SUCCESS)
}

}

编辑:

执行脚本的输入:

 {"pID":"1029409411108724738",
  "contentType":"text",
  "published":"2018-08-14 16:48:23Z",
  "crawled":"2018-08-14 12:48:33-04:00",
  "ID":"765"}

executeScript 的输出:

ops_testQueue,"ID"=765 value=1 1534265314969999870

我错过了什么吗？

Answer 1

我在这里看到了一些事情:

1. 我不知道Nashorn(Java的JS引擎)是否支持完整的lambda 语法，我可以通过将 lambda 设为函数来使其工作 (参见下面的脚本)。
2. 您引用 json 变量来从键获取值，但我认为您需要 message_content。
3. result 未定义，因此在推送时会出现错误。

这是您的脚本的编辑版本，我必须按照您想要的方式工作(但如果我错了，请纠正我):

flowFile = session.get();
if (flowFile != null) {

var StreamCallback = 
Java.type("org.apache.nifi.processor.io.StreamCallback");
var IOUtils = Java.type("org.apache.commons.io.IOUtils");
var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");
var transformed_message = {};
var error = false;
var line = "ops_testQueue";

flowFile = session.write(flowFile, new StreamCallback(function 
  (inputStream, outputStream) {

    var content = IOUtils.toString(inputStream, 
  StandardCharsets.UTF_8); // message or content
    var message_content = {};
    try {
        message_content = JSON.parse(content);
        if(Array.isArray(message_content)){

    }
    var jkey = "";
    Object.keys(message_content).forEach(function(key) {
         var value = message_content[key];
         //result.push(key + '=' + value);
         jkey = "," +  "\"" + key + "\""  + '=' + value   
        });
    line = line + jkey +
            " value=" + "1"
            + " " + Date.now() * 1000000;

        // Write output content
        if (transformed_message) {
            outputStream.write(line.getBytes(StandardCharsets.UTF_8));
        }
    } catch (e) {
        error = true;
        log.error(e);
        outputStream.write(content.getBytes(StandardCharsets.UTF_8));
    }
 }));
 if (transformed_message.post_state) {
    flowFile = session.putAttribute(flowFile, "type", 
    transformed_message.type);
}

if (error) {
    session.transfer(flowFile, REL_FAILURE)
} else {
    session.transfer(flowFile, REL_SUCCESS)
}
}