在Provenance存储库中存储每个FlowFile的历史记录。此历史记录用于提供每个数据的数据沿袭(也称为产销监管链)。每次为FlowFile发生事件(创建,分叉,克隆,修改FlowFile等)时,都会创建一个新的Provenance事件。这个出处事件是流文件的快照,因为它看起来就是在那个时间点存在的流。创建Provenance事件后,它将复制所有FlowFile的属性和指向FlowFile内容的指针,并将其与FlowFile的状态(例如其与其他出处事件的关系)聚合到Provenance存储库里。该快照将不会更改,直到过期。根据“nifi.properties”文件中的指定,Provenance存储库将在完成后的一段时间内保留所有这些来源事件。
本文主要研究一下nifi的AbstractBinlogTableEventWriter
为了创建高效的数据流处理流程,需要了解可用的处理器(Processors )类型,NiFi提供了大约近300个现成的处理器。这些处理器提供了可从不同系统中提取数据,路由,转换,处理,拆分和聚合数据以及将数据分发到多个系统的功能。如果还不能满足需求,还可以自定义处理器。
需求:随机生成一些测试数据集,对生成的数据进行正则匹配,对匹配后的数据进行输出到外部文件中。以上需要用到的“GenerateFlowFile”、“ReplaceText”、“PutFile”处理器。
本文通过Groovy,Jython,Javascript(Nashorn)和JRuby中的代码示例,介绍了有关如何使用Apache NiFi处理器ExecuteScript完成某些任务的各种方法。本文中的内容包括:
NiFi是美国国家安全局开发并使用了8年的可视化数据集成产品,2014年NAS将其贡献给了Apache社区,2015年成为Apache顶级项目
系统正在积极处理的FlowFiles保存在JVM内存中的Hash Map中。这使它们的处理效率非常高,但是由于多种原因,例如断电,内核崩溃,系统升级和维护周期,因此需要一种辅助机制来在整个进程重新启动中提供数据的持久性。FlowFile存储库是系统中当前存在的每个FlowFiles的元数据的Write-Ahead Log(或数据记录)。该FlowFile元数据包括与FlowFile相关联的所有attributes,指向FlowFile实际内容的指针(该内容存在于内容存储库中)以及FlowFile的状态,例如FlowFile所属的Connection/Queue。预写日志为NiFi提供了处理重启和意外系统故障所需的弹性。
NIFI可以处理各种各样的数据源和不同格式的数据。你可以从一个源中获取数据,对其进行转换,然后将其推送到另一个目标存储地。
以上案例用到的处理器有“QueryDatabaseTable”、“ConvertAvroToJSON”、“SplitJson”、“PutHDFS”四个处理器。
这是疯狂的水流。就像您的应用程序处理疯狂的数据流一样。如果您独自完成所有工作,那么很难将数据从一个存储路由到另一个存储,应用验证规则并解决数据治理,大数据生态系统中的可靠性问题。
NiFi Connection是在两个已连接的NiFi处理器组件之间临时保存FlowFiles的位置。每个包含排队的NiFi FlowFiles的Connection在JVM堆中都会占一些空间。本文将对Connection进行分析,探究NiFi如何管理在该Connection中排队的FlowFiles和Connection对堆和性能的影响。
案例:使用NiFi将某个目录下产生的json类型的日志文件导入到Hive。这里首先将数据通过NiFi将Json数据解析属性,然后手动设置数据格式,将数据导入到HDFS中,Hive建立外表映射此路径实现外部数据导入到Hive中。
单独启动“GenerateFlowFile”处理器后,可以观察到对应的Connection连接队列中有数据,在Connection连接上右键“List Queue”可以查看队列中的FlowFile信息:
NiFi的基本设计理念是基于数据流的编程Flow-Based Programming(FBP),应用是由处理器、连接器组成的网络。数据进入一个节点,由该节点对数据进行处理,根据不同的处理结果将数据路由到后续的其他节点进行处理。这是NiFi的流程比较容易可视化的一个原因。以下是NiFi的一些概念:
PutDatabaseRecord处理器使用指定的RecordReader从传入的流文件中读取(可能是多个,说数组也成)记录。这些记录将转换为SQL语句,并作为一个批次执行。如果发生任何错误,则将流文件路由到failure或retry,如果执行成功,则将传入的流文件路由到success。处理器执行的SQL语句类型通过Statement Type属性指定,该属性接受一些硬编码的值,例如INSERT,UPDATE和DELETE,使用“Use statement.type Attribute”可以使处理器获取流文件属性中的语句类型。
DataFlow Manager(DFM)是NiFi用户,具有添加,删除和修改NiFi数据流组件的权限。
现在我们要自定义一个Processor,假设它叫MyProcessor.java,那么这个Java文件写在哪里呢?
注意:以上需要在NiFi集群中的每个节点上创建“/root/test/logdata”文件,“logdata”是文件,而非目录。
以上案例需要用到的处理器有:“CaptureChangeMySQL”、“RouteOnAttribute”、“EvaluateJsonPath”、“ReplaceText”、“PutHiveQL”。
一些处理器支持配置运行持续时间(Run Duration)。此设置告诉处理器在单个任务中继续使用同一task尽可能多地来处理来自传入队列的的FlowFiles(或成批的流文件)。 对于处理单个任务本身非常快并且FlowFile数量也很大的处理器来说,这是一个理想的选择。
该处理器使用Hive流将流文件数据发送到Apache Hive表。传入的流文件需要是Avro格式,表必须存在于Hive中。有关Hive表的需求(格式、分区等),请参阅Hive文档。分区值是根据处理器中指定的分区列的名称,然后从Avro记录中提取的。注意:如果为这个处理器配置了多个并发任务,那么一个线程在任何时候只能写入一个表。写入同一表的其他任务将等待当前任务完成对表的写入。
简介:本文主要讲解Apache NIFI的调度策略,对象主要是针对Processor组件。本文假定读者已经对Apache NIFI有了一定的了解和使用经验,同时作者也尽可能的去讲解的更透彻,使得本文尽可能让对NIFI接触不深的读者也能够看懂。
在下面的列表中,必需属性的名称以粗体显示。任何其他属性(不是粗体)都被认为是可选的,并且指出属性默认值(如果有默认值),以及属性是否支持表达式语言。
内容存储库就是本地存储所有FlowFiles内容的地方,通常是三个存储库中最大的。该存储库利用不变性和写时复制来最大提升读写速度和保证线程安全性。Content Repo的核心设计是将FlowFile的内容保存在磁盘上,并仅在需要时才将其读入JVM内存。这使NiFi可以处理大量小的对象,而无需生产者和消费者处理器将完整的对象保存在内存中。因此,在不损害内存的情况下,非常容易执行诸如拆分,聚合和转换非常大的对象之类的操作。
NiFi对其摄取的每个数据保存明细。当数据通过系统处理并被转换,路由,拆分,聚合和分发到其他端点时,这些信息都存储在NiFi的Provenance Repository中。为了搜索和查看此信息,我们可以从全局菜单中选择数据源(Data Provenance),也可以在对应的处理器上右键选择“View data provenance”进行查看。
该处理器根据配置将二进制编码的Avro数据文件分割成更小的文件。输出策略决定split后的文件是Avro数据文件,还是只保留Avro记录(在FlowFile属性中包含元数据信息 )。输出总是二进制编码的。
Fayson在前面的文章介绍了什么是NiFi,参考《0622-什么是Apache NiFi》。同时对如何在CDH中使用Parcel安装CFM做了介绍,参考《0623-6.2.0-如何在CDH中安装CFM》。本文会首先对NiFi的使用做一下简单的介绍,然后对处理器(Processor)进行详细介绍。
在本实验中,您将运行一个简单的 Python 脚本来模拟来自一些假设的机器的 IoT 传感器数据,并将数据发送到 MQTT 代理 ( mosquitto )。MQTT 代理扮演网关的角色,通过“mqtt”协议连接到许多不同类型的传感器。您的集群附带模拟脚本发布到的嵌入式 MQTT 代理。为方便起见,我们将使用 NiFi 来运行脚本而不是 Shell 命令。
简单地说,NiFi就是为了实现系统间数据流的自动化而构建的。虽然术语“数据流”用于各种上下文,但我们在此处使用它来表示系统之间的自动和管理信息流。这个问题空间一直存在,因为企业有多个系统,其中一些系统创建数据,一些系统消耗数据。已经讨论并广泛阐述了出现的问题和解决方案模式。企业集成模式[eip]中提供了一个全面且易于使用的表单。
2006年NiFi由美国国家安全局(NSA)的Joe Witt创建。2015年7月20日,Apache 基金会宣布Apache NiFi顺利孵化成为Apache的顶级项目之一。NiFi初始的项目名称是Niagarafiles,当NiFi项目开源之后,一些早先在NSA的开发者们创立了初创公司Onyara,Onyara随之继续NiFi项目的开发并提供相关的支持。Hortonworks公司收购了Onyara并将其开发者整合到自己的团队中,形成HDF(Hortonworks Data Flow)平台。2018年Cloudera与Hortonworks合并后,新的CDH整合HDF,改名为Cloudera Data Flow(CDF),并且在最新的CDH6.2中直接打包,参考《0603-Cloudera Flow Management和Cloudera Edge Management正式发布》,而Apache NiFi就是CFM的核心组件。
许多第一次接触使用NIFI的同学在同步关系型数据库的某一张表的时候,可能会拖拽出类似于下面的一个流程。
NiFi使用预写日志来跟踪FlowFiles(即数据记录)在系统中流动时的变化。该预写日志跟踪FlowFiles本身的更改,例如FlowFile的属性(组成元数据的键/值对)及其状态,再比如FlowFile所属的Connection /Queue。
在组件工具栏下的NiFi屏幕顶部附近有一个条形,称为状态栏。它包含一些关于NiFi当前健康状况的重要统计数据:活动线程的数量可以指示NiFi当前的工作状态,排队统计数据表示当前在整个流程中排队的FlowFile数量以及这些FlowFiles的总大小。
Apache NiFi是什么?NiFi官网给出如下解释:“一个易用、强大、可靠的数据处理与分发系统”。通俗的来说,即Apache NiFi 是一个易于使用、功能强大而且可靠的数据处理和分发系统,其为数据流设计,它支持高度可配置的指示图的数据路由、转换和系统中介逻辑。 为了对NiFi能够表述的更为清楚,下面通过NiFi的架构来做简要介绍,如下图所示。
列名转换是ETL过程中常常遇到的场景。例如来源表user的主键id,要求写入目标表user的uid字段内,那么就需要列名转换.
流量控制器是操作的大脑。它为运行扩展提供线程,并管理扩展何时接收要执行的资源的时间表。
nifi.properties文件中有三个属性涉及 NiFi 内容存储库中内容的存档。
开始之前,看一下源码结构,nifi的注解都是在nifi-api moudle中的。
通常我们在NIFI里最常见的使用场景就是读写关系型数据库,一些组件比如GenerateTableFetch、ExecuteSQL、PutSQL、ExecuteSQLRecord、PutDatabaseRecord等等,都会有一个属性配置大概叫Database Connection Pooling Service的,对应的接口是DBCPService,其实现类有:HiveConnectionPool DBCPConnectionPool DBCPConnectionPoolLookup。我们用的最多的就是DBCPConnectionPool。具体怎么配置这里就不赘述了,看对应的Controller Service文档就可以了。
本教程涵盖了Apache NiFi的核心概念及其在其中流量管理,易用性,安全性,可扩展架构和灵活扩展模型非常重要的环境中所扮演的角色。
该处理器用于生成在表中执行分页查询的SQL 查询语句,分区(属性partition)大小以及表的行数决定页面的大小和数量以及生成的流文件。此外,可以通过设置最大值列来实现增量抓取数据,处理器会跟踪列的最大值,从而只抓取列值超过已记录到的最大值的行,该处理器只在主节点上运行,可以接受传入的连接;
当客户希望在生产环境中使用NiFi时,这些通常是第一个提出的问题。他们想知道他们将需要多少硬件,以及NiFi是否可以容纳其数据速率。
在本次实验中,您将实施一个数据管道来处理之前从边缘捕获的数据。您将使用 NiFi 将这些数据摄取到 Kafka,然后使用来自 Kafka 的数据并将其写入 Kudu 表。
克隆,即复制一个对象,该对象的属性与被复制的对象一致,如果不使用Object类中的clone方法实现克隆,可以自己new出一个对象,并对相应的属性进行数据,这样也能实现克隆的目的。
实现一个深克隆是面试中常见的问题的,可是绝大多数面试者的答案都是不完整的,甚至是错误的,这个时候面试官会不断追问,看看你到底理解不理解深克隆的原理,很多情况下一些一知半解的面试者就原形毕漏了.
本文实例讲述了PHP设计模式之原型模式定义与用法。分享给大家供大家参考,具体如下:
领取专属 10元无门槛券
手把手带您无忧上云