【NIFI】 Apache NiFI 之 ExecuteScript处理（一）

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

本例介绍NiFI ExecuteScript处理器的使用，使用的脚本引擎ECMScript

FlowFile I / O简介

NiFi中的流文件由两个主要组件构成，即属性和内容。属性是关于内容/流文件的元数据，我们在本系列的第1部分中看到了如何使用ExecuteScript来操作它们。流文件的内容本质上只是一个字节集合，没有固有的结构，模式，格式等。各种NiFi处理器假设传入的流文件具有特定的模式/格式（或者从属性中确定它作为“mime.type”或以其他方式推断它。然后，这些处理器可以基于文件确实具有该格式的假设来对内容起作用（并且如果它们不这样，则经常转移到“失败”关系）。处理器也可以输出指定格式的流文件，这在处理器中有描述。NiFi文档。

流文件内容的输入和输出（I / O）通过ProcessSession API提供，因此是ExecuteScript的“session”变量（有关更多信息，请参阅第1部分）。一种机制是将回调对象传递给对session.read（）或session.write（）的调用。将为FlowFile对象创建InputStream和/或OutputStream，并使用相应的回调接口调用回调对象，并传入InputStream和/或OutputStream引用以供回调使用。有三个主要的回调接口，每个接口都有自己的用例：

InputStreamCallback

session.read( flowFile ， inputStreamCallback )方法使用此接口提供InputStream，从中读取流文件的内容。界面有一个方法：

 void process(InputStream in) throws IOException

此接口提供托管输入流以供使用。虽然可以手动关闭流，但输入流会自动打开和关闭。如果您只是从特定的流文件中读取而不是写回来，那么这是您将使用的表单。

例如，当您想要处理传入的流文件，但创建许多输出流文件时，如 SplitText处理器。

OutputStreamCallback

session.write( flowFile ， outputStreamCallback )方法使用此接口来提供 要写入流 文件内容的OutputStream。界面有一个方法：

 void process(OutputStream out) throws IOException

此接口提供托管输出流以供使用。尽管可以手动关闭流，但输出流会自动打开和关闭 - 如果包含这些流的任何流打开应该清除的资源，则非常重要。

例如，ExecuteScript将从内部或外部文件生成数据，但不生成流文件。然后你将使用session.create（）创建一个新的FlowFile，然后使用session.write( flowFile ， outputStreamCallback) 来插入内容。

StreamCallback

session.write（ flowFile ， streamCallback ）方法使用此接口来提供InputStream和OutputStream，从中读取和/或写入流文件的内容。界面有一个方法：

 void process(InputStream in, OutputStream out) throws IOException

此接口提供托管输入和输出流以供使用。虽然可以手动关闭流，但输入流会自动打开和关闭 - 如果包含这些流的任何流打开应该清除的资源，则非常重要。

例如，当您想要处理传入的流文件并用新的东西覆盖其内容时，例如 EncryptContent处理器。

由于这些回调是Java对象，因此脚本必须创建一个并将其传递给会话方法，还有其他读取和写入流文件的方法，包括：

- 使用session.read（ flowFile ）返回一个InputStream。这减轻了对InputStreamCallback的需求，而是返回可以读取的InputStream。作为交换，您必须手动管理（关闭，例如）InputStream。

- 使用session.importFrom（ inputStream ， flowFile ）从InputStream写入FlowFile。这取代了传递了OutputStreamCallback的session.write（）的需要。

ExecuteScript介绍

ExecuteScript是一个多功能处理器，允许用户使用编程语言编写自定义逻辑，每次触发ExecuteScript处理器时都会执行该编程语言。为脚本提供以下变量绑定以启用对NiFi组件的访问：

session ：这是对分配给处理器的ProcessSession的引用。该会话允许您对流文件执行操作，如create（），putAttribute（）和transfer（），以及read（）和write（）。

context ：这是对处理器的ProcessContext的引用。它可用于检索处理器属性，关系，Controller Services和StateManager。

log ：这是对处理器的ComponentLog的引用。使用它将消息记录到NiFi，例如log.info（'Hello world！'）

REL_SUCCESS ：这是对为处理器定义的“成功”关系的引用。它也可以通过引用父类的静态成员（ExecuteScript）来继承，但是某些引擎（如Lua）不允许引用静态成员，因此这是一个便利变量。它还节省了必须使用关系的完全限定名称。

REL_FAILURE ：这是对为处理器定义的“失败”关系的引用。与REL_SUCCESS一样，它也可以通过引用父类的静态成员（ExecuteScript）来继承，但是某些引擎（如Lua）不允许引用静态成员，因此这是一个便利变量。它还节省了必须使用关系的完全限定名称。

动态属性 ：ExecuteScript中定义的任何动态属性都将作为设置为与动态属性对应的PropertyValue对象的变量传递给脚本引擎。这允许您获取属性的String值，还可以根据NiFi表达式语言评估属性，将值转换为适当的数据类型（例如布尔值等）等。因为动态属性名称变为脚本的变量名，您必须知道所选脚本引擎的变量命名属性

ExecuteScript使用

1、从会话中获取传入的流文件

目的：有到ExecuteScript的传入连接，并希望从队列中检索一个流文件以进行处理

方法：使用会话对象中的get()方法。此方法返回要处理的下一个最高优先级FlowFile的FlowFile。如果没有要处理的FlowFile，则该方法将返回null。请注意，即使FlowFiles有稳定的流入处理器，也可能返回null。如果处理器有多个并发任务，并且其他任务已经检索到FlowFiles，则会发生这种情况。如果脚本需要FlowFile继续处理，那么如果从session.get()返回null，它应立即返回

Examples

Javascript

 var flowFile = session.get();

 if (flowFile != null) {

     // All processing code goes here

 }

2、从会话中获取多个传入流文件

目的：有到ExecuteScript的传入连接，并希望从队列中检索多个流文件以进行处理

方法：使用会话对象中的get（ maxResults ）方法。此方法从工作队列返回 maxResults FlowFiles。如果没有可用的FlowFiles，则返回一个空列表（该方法不返回null）。注意：如果存在多个传入队列，则根据是否在单个调用中轮询所有队列或仅轮询单个队列，未指定行为。话虽如此，这里描述了观察到的行为（对于NiFi 1.1.0+和之前）

Examples

Javascript

 flowFileList = session.get(100)

 if(!flowFileList.isEmpty()) {

   for each (var flowFile in flowFileList) {

        // Process each FlowFile here

   }

 }

3、创建一个新的FlowFile

目的：生成一个新的FlowFile以发送到下一个处理器

方法：使用会话对象中的create()方法。此方法返回一个新的FlowFile对象，您可以对其执行进一步处理

Examples

Javascript

 var flowFile = session.create();

 // Additional processing here

4、从父FlowFile创建新的FlowFile

目的：希望基于传入的FlowFile生成新的FlowFile

方法：使用会话对象中的create（ parentFlowFile ）方法。此方法采用父FlowFile引用并返回新的子FlowFile对象。新创建的FlowFile将继承除UUID之外的所有父属性。此方法将自动生成Provenance FORK事件或Provenance JOIN事件，具体取决于在提交ProcessSession之前是否从同一父级生成其他FlowFiles

Examples

Javascript

 var flowFile = session.get();

 if (flowFile != null) {

     var newFlowFile = session.create(flowFile);

     // Additional processing here

 }

5、向流文件添加属性

目的：有一个要添加自定义属性的流文件

方法：使用会话对象中的putAttribute（ flowFile ， attributeKey ， attributeValue ）方法。此方法使用给定的键/值对更新给定的FlowFile属性。注意：“uuid”属性对于FlowFile是固定的，不能修改; 如果密钥名为“uuid”，则将被忽略。

Examples

Javascript

 var flowFile = session.get();

 if (flowFile != null) {

     flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')

 }

6、向流文件添加多个属性

目的：有一个要添加自定义属性的流文件

方法：使用会话对象中的putAllAttributes（ flowFile ， attributeMap ）方法。此方法使用给定Map中的键/值对更新给定的FlowFile属性。注意：“uuid”属性对于FlowFile是固定的，不能修改; 如果密钥名为“uuid”，则将被忽略。

Examples

Javascript

 var number2 = 2;

 var attrMap = {'myAttr1':'1', 'myAttr2': number2.toString()}

 var flowFile = session.get()

 if (flowFile != null) {

     flowFile = session.putAllAttributes(flowFile, attrMap)

 }

7、从流文件中获取属性

目的：有一个流文件，您可以从中检查属性

方法：使用FlowFile对象中的getAttribute（ attributeKey ）方法。此方法返回给定attributeKey的String值，如果未找到attributeKey，则返回null。这些示例显示了“filename”属性值的检索。

Examples

Javascript

 var flowFile = session.get()

 if (flowFile != null) {

     var myAttr = flowFile.getAttribute('filename')

 }

 var flowFile = session.get()

 if (flowFile != null) {

     var attrs = flowFile.getAttributes();

     for each (var attrKey in attrs.keySet()) {

        // Do something with attrKey (the key) and/or attrs[attrKey] (the value)

   }

 }

8、将流文件传输到关系

目的：处理流文件（新文件或传入文件）后，您希望将流文件传输到关系（“成功”或“失败”）。在这个简单的情况下，让我们假设有一个名为“errorOccurred”的变量，它指示FlowFile应该传输到哪个关系。

方法：使用会话对象中的transfer（ flowFile ， relationship ）方法。从文档中：此方法根据给定的关系将给定的FlowFile传输到适当的目标处理器工作队列。如果关系导致多个目标，则复制FlowFile的状态，使得每个目标都接收FlowFile的精确副本，尽管每个目标都具有其自己的唯一标识。

注意：ExecuteScript将在每次执行结束时执行session.commit()以确保已提交操作。您不需要（也不应该）在脚本中执行session.commit()。

Examples

Javascript

 var flowFile = session.get();

 if (flowFile != null) {

    // All processing code goes here

    if(errorOccurred) {

      session.transfer(flowFile, REL_FAILURE)

    }

    else {

      session.transfer(flowFile, REL_SUCCESS)

    }

 }

9、以指定的日志记录级别向日志发送消息

目的：将处理期间发生的某些事件报告给日志记录框架。

方法：将log变量与warn()，trace()，debug()，info()或error()方法一起使用。这些方法可以使用单个String，或者后跟对象数组的String，或者后跟对象数组后跟Throwable的String。第一个用于简单消息。当您有一些要记录的动态对象/值时，将使用第二个。要在消息字符串中引用这些，请在消息中使用“{}”。这些是按照外观的顺序针对Object数组进行评估的，因此如果消息显示为“Found these things：{} {} {}”并且Object数组为['Hello'，1，true]，则记录的消息将为“找到这些东西：你好1真的”。这些日志记录方法的第三种形式也采用Throwable参数
Examples

Javascript

 var ObjectArrayType = Java.type("java.lang.Object[]");

 var objArray = new ObjectArrayType(3);

 objArray[0] = 'Hello';

 objArray[1] = 1;

 objArray[2] = true;

 log.info('Found these things: {} {} {}', objArray)

10、使用回调读取传入流文件的内容

目的：有到ExecuteScript的传入连接，并希望从队列中检索流文件的内容以进行处理

方法：使用read( flowFile ， inputStreamCallback) 来自会话对象的方法。传入read（）方法需要一个InputStreamCallback对象。请注意，因为InputStreamCallback是一个对象，所以默认情况下内容只对该对象可见。如果需要使用read（）方法之外的数据，请使用更全局范围的变量。这些示例将传入流文件的完整内容存储到String中（使用Apache Commons的IOUtils类）。注意：对于大流量文件，这不是最好的技术; 相反，您应该只读取您需要的数据，并根据需要进行处理。对于像SplitText这样的东西，你可以一次读取一行并在InputStreamCallback中处理它，或者使用前面提到的session.read（flowFile）方法来获得在回调之外使用的InputStream引用。

Examples

Javascript

 var InputStreamCallback =  Java.type("org.apache.nifi.processor.io.InputStreamCallback")

 var IOUtils = Java.type("org.apache.commons.io.IOUtils")

 var StandardCharsets = Java.type("java.nio.charset.StandardCharsets")

 var flowFile = session.get();

 if(flowFile != null) {

   // Create a new InputStreamCallback, passing in a function to define the interface method

   session.read(flowFile,

     new InputStreamCallback(function(inputStream) {

         var text = IOUtils.toString(inputStream, StandardCharsets.UTF_8);

         // Do something with text here

     }));

 }

11、使用回调将内容写入传出流文件

目的：传出流文件生成内容

方法：使用会话对象中的write( flowFile ， outputStreamCallback) 方法。传递给write()方法需要一个OutputStreamCallback对象。请注意，因为OutputStreamCallback是一个对象，所以默认情况下内容只对该对象可见。如果需要使用write()方法之外的数据，请使用更全局范围的变量。这些示例将示例String写入flowFile。

Examples

Javascript

 var OutputStreamCallback =  Java.type("org.apache.nifi.processor.io.OutputStreamCallback");

 var IOUtils = Java.type("org.apache.commons.io.IOUtils");

 var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");

 var flowFile = session.get();

 if(flowFile != null) {

   // Create a new OutputStreamCallback, passing in a function to define the interface method

   flowFile = session.write(flowFile,

     new OutputStreamCallback(function(outputStream) {

         outputStream.write("Hello World!".getBytes(StandardCharsets.UTF_8))

     }));

 }

12、使用回调覆盖带有更新内容的传入流文件

目的：重用传入的流文件，但希望修改其传出流文件的内容。

方法：使用write( flowFile ， streamCallback) 来自会话对象的方法。传递给write()方法需要StreamCallback对象。StreamCallback提供InputStream（来自传入流文件）和outputStream（用于该流文件的下一个版本），因此您可以使用InputStream获取流文件的当前内容，然后修改它们并将它们写回到流文件。这会覆盖流文件的内容，因此对于追加，您必须通过附加到读入内容来处理它，或者使用不同的方法（使用session.append()而不是session.write()）。请注意，由于StreamCallback是一个对象，因此默认情况下内容仅对该对象可见。如果需要使用write()方法之外的数据，请使用更全局范围的变量。

Examples

Javascript

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 flowFile = session.get();

if(flowFile != null) {

  // Create a new StreamCallback, passing in a function to define the interface method

  flowFile = session.write(flowFile,

    new StreamCallback(function(inputStream, outputStream) {

        var text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)

        outputStream.write(text.split("").reverse().join("").getBytes(StandardCharsets.UTF_8))

    }));

}

13、处理脚本处理过程中的错误

目的：脚本中发生错误（通过数据验证或抛出异常），并且您希望脚本正常处理它。

方法：对于异常，使用脚本语言的异常处理机制（通常它们是try / catch块）。对于数据验证，您可以使用类似的方法，但定义一个布尔变量，如“valid”和if / else子句而不是try / catch子句。ExecuteScript定义“成功”和“失败”关系; 通常，您的处理将“好”流文件转移到成功，“坏”流文件转换为失败（在后一种情况下记录错误）

Examples

Javascript

 var flowFile = session.get();

 if(flowFile != null) {

   try {

     // Something that might throw an exception here

     // Last operation is transfer to success (failures handled in the catch block)

     session.transfer(flowFile, REL_SUCCESS)

 } catch(e) {

   log.error('Something went wrong', e)

   session.transfer(flowFile, REL_FAILURE)

 }

 }

ExecuteScript-Demo

1、页面如下图

2、GenerateFlowFile

2、ExecuteScript

脚本内容：

 var InputStreamCallback =  Java.type("org.apache.nifi.processor.io.InputStreamCallback");

 var OutputStreamCallback =  Java.type("org.apache.nifi.processor.io.OutputStreamCallback");

 var IOUtils = Java.type("org.apache.commons.io.IOUtils");

 var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");

 var flowFile = session.get();

 if(flowFile != null) {

     try {

         var text = "";

         // 读取flowFile中内容

         session.read(flowFile,new InputStreamCallback(function(inputStream) {

             var str = IOUtils.toString(inputStream, StandardCharsets.UTF_8);

             //由JSON字符串转换为JSON对象

             var obj = JSON.parse(str);

             obj.age = 18

             //将JSON对象转化为JSON字符

             text = JSON.stringify(obj); 

         }));

         // 向flowFile中写入内容

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

             outputStream.write(text.getBytes(StandardCharsets.UTF_8))

         }));

         session.transfer(flowFile, REL_SUCCESS)

     } catch(e) {

         log.error('Something went wrong', e)

         session.transfer(flowFile, REL_FAILURE)

     }

 }

3、PutFile