标签: apache-nifi

我打算做一个类项目,并且经历了一些技术,我可以自动化或设置系统之间的数据流,并发现它们有几个,即Apache NiFi和StreamSets(据我所知).我无法理解的是它们与可以使用它们的用例之间的区别？我是新手,如果有人能解释我,我会非常感激.谢谢

我只是想知道迷你NiFi MiNiFi与NiFi有何不同？

我们还有其他功能吗？为什么需要引入它？

任何想法或链接都会有很大的帮助.

Airflow和Nifi在工作流程中执行相同的工作吗？每个人的赞成/赞成是什么？我需要读取一些json文件,向其添加更多自定义元数据并将其放入要处理的Kafka队列中.我能够在Nifi中做到这一点.我还在研究Airflow.我正在尝试为我的项目选择最好的工作流引擎谢谢!

我对Nifi及其功能以及相应的用例有疑问.

我已经读到Nifi真正的目标是创建一个允许基于流程处理的空间.在玩了一下Nifi之后,我也意识到它能够以对我有用的方式建模/塑造数据.可以公平地说Nifi也可以用于数据建模吗？

谢谢!

我正在使用Groovy脚本上的Apache NiFi 0.5.1来将传入的Json值替换为映射文件中包含的值.映射文件看起来像这样(它是一个简单的.txt):

Header1;Header2;Header3
 A;some text;A2

我从以下开始:

import groovy.json.JsonBuilder 
import groovy.json.JsonSlurper 
import java.nio.charset.StandardCharsets 

def flowFile = session.get(); 
if (flowFile == null) { 
    return; 
} 

flowFile = session.write(flowFile, 
        { inputStream, outputStream -> 

            def content = """ 
{ 
  "field1": "A"
  "field2": "A", 
  "field3": "A" 

}""" 

            def slurped = new JsonSlurper().parseText(content) 
            def builder = new JsonBuilder(slurped) 
            builder.content.field1 = "A"
            builder.content.field2 = "some text" 
            builder.content.field3 = "A2" 
            outputStream.write(builder.toPrettyString().getBytes(StandardCharsets.UTF_8)) 
        } as StreamCallback) 
session.transfer(flowFile, ExecuteScript.REL_SUCCESS)

这第一步工作得很好,虽然它是硬编码的,但远非理想.我最初的想法是使用ReplaceTextWithMapping来执行替换,但是它不适用于复杂的映射文件(例如多列).我想更进一步,但我不知道如何去做.首先,我想读取传入的流文件,而不是传入整个编码的JSON.这怎么可能在NiFi？在将脚本作为ExecuteScript的一部分运行之前,我通过UpdateAttribute输出带有内容的.Json文件,其中filename = myResultingJSON.json.此外,我知道如何使用Groovy(String mappingContent= new File('/path/to/file').getText('UTF-8')加载.txt文件,但是如何使用加载的文件来执行替换,以便我生成的JSON看起来像这样:

{ 
  "field1": "A" …

我是Nifi和python的新手

我想执行我的python脚本.所以使用了ExecuteScript并尝试导入某些模块.我这样导入:

import json, sftp, paramiko

虽然我安装了sftp,但是当我在Executescript中导入它时,它会显示"无法处理会话.在第1行没有名为sftp的模块"

which -a sftp
/usr/bin/sftp

当导入paramiko时,也得到了同样的错误.

我刚刚开始使用NiFi,当第一次拍摄时它很棒.

当事情失败时,我想从有问题的处理器中查看流文件或输出(stdout + stderr).是否有一种简单的方法可以转储所有流文件的属性,或只是浏览流文件？

我觉得我错过了一些非常明显的东西,但是大多数教程只是走过一切都是第一次拍摄的例子.

我意识到,正如他们的文档所定义的那样,NiFi "生产中会持续改进".因此,这不适合用作传统的开发工具.然而,对于我正在研究的项目,已经决定这是我们将要使用的工具,所以我宁愿不讨论这个的优点,因为我意识到会有一些问题.

例如,如果我将更改推送到现有环境(从登台到生产)并且目标中有实时编辑,它们将被覆盖.所以我对如何组织开发生命周期有疑问.

• 是否可以合并多个开发人员并行完成的更改(合并导出的xml模板文件)？我猜合并任何重大变化可能很困难,但没有尝试过.
• 如何管理版本更改？我假设您可以将整个配置导出为模板并将其检入版本控制中？
• 如何将流部署到其他服务器？您是否可以部署一个库存NiFi部署,然后使用NiFi REST API从导出的模板(如上所述)更新它？
• 如何管理部署到可能具有不同配置的不同环境？你需要更新模板XML文件吗？或者我可以动态地从Zookeeper这样的东西中提取它吗？

Apache Beam和Apache Nifi有哪些用例？它们似乎都是数据流引擎.如果两者都有相似的用例,哪两个更好？

在尽力找到与此问题相关的先前问题和示例之后,仍然没有找到我正在寻找的答案,我想我会自己提交一个问题.

由于以下原因,ExecuteStreamCommand对我来说似乎是完美的处理器:

• 我能够执行任何Python脚本并避免使用Jython(与ExecuteScript类似).Jython对我来说不是一个选择.
• 我可以接受FlowFiles.这是必要的,因为我的脚本使用前一个处理器的输出.此外,我喜欢将数据保存在"NiFi管理"下的想法.
• 它写了一个"执行状态",它对路由很有用.

简而言之,我正在尝试使用ExecuteStreamCommand:

• 摄取先前处理器的输出(Scrapy蜘蛛输出带有JSON行的文本文件)
• 调用python脚本(例如python3 my_script.py)
• 加载我的python脚本中提取的FlowFile.
• 选择FlowFile的内容.
• 在python中操作FlowFile的内容.
• 输出原始FlowFile的更新版本或创建新版本.
• 使用更新/新的FlowFile继续我的NiFi流程.

为清楚起见,我目前不明白:

• 如何调用python脚本(来自ExecuteStreamCommand Processor)
• 如何从Python中加载FlowFile
• 如何在Python中更新或创建新的FlowFile
• 如何将更新后的FlowFile从Python输出回NiFi.

我遇到过ExecuteScript的各种示例,但不幸的是,这些并不完全转化为使用ExecuteStreamCommand.

先感谢您.任何建议表示赞赏.