NiFi中执行远程命令：ExecuteScript,Groovy,和Sshoogr

?

Apache NiFi 已经有一些远程命令的执行器，如 ExecuteProcess 和 ExecuteStreamCommand。可以通过操作系统的ssh命令远程执行脚本，还可以带上不同的参数。通常，这通过一个基于bash脚本执行器来执行，可以带多个ssh命令。

准备工作

另外一种方法，可以使用 ExecuteScript 或者 InvokeScriptedProcessor,通过第三方库通过ssh去执行远程命令。本文中,我使用 Groovy 作为脚本语言并且使用 Sshoogr,Sshoogr用于Groovy的 SSH DSL。

Sshoogr 可以通过 SDK Manager 安装或下载 (例如从 Maven Central ). 关键需要所有的支持库 (包括Sshoogr 和依赖库)在同一个目录中。将 ExecuteScript's Module Directory 属性指向该目录,然后所有的 JARs 都将能够在 Groovy script中使用。

为了使这个例子更加灵活，我采用了一些常规的做法。首先,将要执行的commands放在流文件的每一行中。然后,配置参数 (如hostname,username,password,以及 port) 作为动态属性添加到ExecuteScript。 注意，password 属性在这里大小写不敏感; 如果愿意,使用InvokeScriptedProcessor并且明确地添加这个属性 (指明 password 大小写敏感).

为了 Sshoogr 能够工作 (至少在这个例子中),需要远端节点的 RSA key 配置在NiFi 用户的 ~/.ssh/known_hosts 文件中。以为操作系统的差异,有些情况下 SSH 连接将失败，主要因为严格的 host key 检查,因此在 script 我们将在 Sshoogr中关闭这个功能。

输入文件

回到script. 我启动一个标准的代码来获取输入的 flow file,然后使用session.read() 获得每一行，保存到一个命令的列表对象中:

def flowFile = session.get()
if(!flowFile) return

def commands = []
// Read in one command per line
session.read(flowFile,{inputStream ->
    inputStream.eachLine { line -> commands << line  }
} as InputStreamCallback)

然后关闭严格host key 检查:

options.trustUnknownHosts = true

执行脚本

然后使用 Sshoogr DSL 去执行命令（上一步中，从 flow file读取的列表）:

def result = null
remoteSession {

   host = sshHostname.value
   username = sshUsername.value
   password = sshPassword.value
   port = Integer.parseInt(sshPort.value)
        
   result = exec(showOutput: true,failOnError: false,command: commands.join('n'))
}

这里使用前述的 ExecuteScript 动态属性 (sshHostname,sshUsername,sshPassword,sshPort)。后面,有一个截屏显示被加到了ExecuteScript的 配置之中.

这个 exec() 方法是Sshoogr的强大方法. 这里希望命令表的输出放到outgoing的flowfile. 设置"failOnError" 为 false以让所有的命令都会尝试执行,但你也许希望设置为 true，这样命令失败时后续的将不再执行. 这个最终的参数是"command". 如果你不使用 "named-parameter" (aka Map) 的exec() 版本， 而只是执行命令 commands (如，不设置 "showOutput" ),那么 exec() 将接收一个集合:

exec(commands)

在我们,? exec() 方法转换"command" 参数为字符串,替代List "commands",将其转换为换行符分割的字符串，使用 join('n')进行转换。

script 的下一部分是复写传入的 flow file,如果我没有创建一个新的. 如果使用ExecuteScript 创建了一个新的 flow file,确保移除原来的. 如果使用 InvokeScriptedProcessor,你需要定义 "original" 关系并且路由原来的 incoming flow file 到新的。

flowFile = session.write(flowFile,{ outputStream ->
    result?.output?.eachLine { line ->
       outputStream.write(line.bytes)
       outputStream.write('n'.bytes)
    }
} as OutputStreamCallback)

最后,使用 result.exitStatus,我们确认下路由到 flow file 是成功还是失败:

session.transfer(flowFile,result?.exitStatus ? REL_FAILURE : REL_SUCCESS)

全部脚本

这个例子中全部的 script 如下:

import static com.aestasit.infrastructure.ssh.DefaultSsh.*

def flowFile = session.get()
if(!flowFile) return

def commands = []
// Read in one command per line
session.read(flowFile,{inputStream ->
    inputStream.eachLine { line -> commands << line  }
} as InputStreamCallback)

options.trustUnknownHosts = true
def result = null
remoteSession {

   host = sshHostname.value
   username = sshUsername.value
   password = sshPassword.value
   port = Integer.parseInt(sshPort.value)
        
   result = exec(showOutput: true,command: commands.join('n'))
}

flowFile = session.write(flowFile,{ outputStream ->
    result?.output?.eachLine { line ->
       outputStream.write(line.bytes)
       outputStream.write('n'.bytes)
    }
} as OutputStreamCallback)

session.transfer(flowFile,result?.exitStatus ? REL_FAILURE : REL_SUCCESS)

搬到 ExecuteScript 的配置中:

这里我连接到 Hortonworks Data Platform (HDP) Sandbox来运行Hadoop commands,缺省的设置如上所示，我将 script 黏贴进 Script Body,将 Module Directory 指向我的Sshoogr下载目录. 我添加这些到一个简单数据流，该数据流创建了一个 flow file,填充 commands,在远端执行,然后从session中返回的output输出到 logs :

这个例子中，我运行下面的两个命令:

hadoop fs -ls /user
echo "Hello World!" > here_i_am.txt

在 log中 (after LogAttribute runs with payload included),看见输出如下:

在sandbox，我看见文件已被创建:

如你所见，本文描述了使用 ExecuteScript 采用 Groovy 和 Sshoogr 去执行存储在输入流文件中的远程命令. 我创建了 Gist of the template,不过这是用 Apache NiFi 1.0的beta版本创建的，有可能无法在其他版本中使用(如0.x). 但这里已经包含了所有的内容，你可以容易地创建自己的数据流。

原文参考：

http://funnifi.blogspot.jp/2016/08/executing-remote-commands-in-nifi-with.html