在本系列的前一篇博客《将流转化为数据产品》中，我们谈到了减少数据生成/摄取之间的延迟以及从这些数据中产生分析结果和洞察力的日益增长的需求。我们讨论了如何使用带有 Apache Kafka 和 Apache Flink 的Cloudera 流处理(CSA) 来实时和大规模地处理这些数据。在这篇博客中，我们将展示一个真实的例子来说明如何做到这一点，看看我们如何使用 CSP 来执行实时欺诈检测。
在本系列的前一篇博客“将流转化为数据产品”中，我们谈到了减少数据生成/摄取之间的延迟以及从这些数据中产生分析结果和洞察力的日益增长的需求。我们讨论了如何使用带有 Apache Kafka 和 Apache Flink 的Cloudera 流处理(CSP) 来实时和大规模地处理这些数据。在这篇博客中，我们将展示一个真实的例子来说明如何做到这一点，看看我们如何使用 CSP 来执行实时欺诈检测。
NiFi是美国国家安全局开发并使用了8年的可视化数据集成产品，2014年NAS将其贡献给了Apache社区，2015年成为Apache顶级项目
当客户希望在生产环境中使用NiFi时，这些通常是第一个提出的问题。他们想知道他们将需要多少硬件，以及NiFi是否可以容纳其数据速率。
这是疯狂的水流。就像您的应用程序处理疯狂的数据流一样。如果您独自完成所有工作，那么很难将数据从一个存储路由到另一个存储，应用验证规则并解决数据治理，大数据生态系统中的可靠性问题。
使用正确的工具，您可以在不到一小时的时间内构建这样的系统！在此博客文章中，我将向您展示如何使用Raspberry Pi硬件和开源软件（MQTT代理、Apache NiFi、MiNiFi和MiNiFi C2 Server）实现高级IIoT原型。我将专注于体系结构，连接性，数据收集和自动重新配置。
开始之前，看一下源码结构，nifi的注解都是在nifi-api moudle中的。
回顾2020年，Apache NIFI一共发布了7个版本1.12.1、1.12.0、1.11.4、1.11.3、1.11.2、1.11.1、1.11.0。版本发布之频繁前所未有，可以看出NIFI的开源社区贡献力量壮大了许多，同时也更加期待NIFI未来能够给我们带来更多的惊喜。
Fayson在前面的文章介绍了什么是NiFi，参考《0622-什么是Apache NiFi》。同时对如何在CDH中使用Parcel安装CFM做了介绍，参考《0623-6.2.0-如何在CDH中安装CFM》。也介绍过NiFi处理器以及实操，参考《0624-6.2.0-NiFi处理器介绍与实操》。本文会完成第一个NiFi例子，通过NiFi监控一个本地数据目录，定时将新文件put到HDFS。
NIFI可以处理各种各样的数据源和不同格式的数据。你可以从一个源中获取数据，对其进行转换，然后将其推送到另一个目标存储地。
在本实验中，您将运行一个简单的 Python 脚本来模拟来自一些假设的机器的 IoT 传感器数据，并将数据发送到 MQTT 代理 ( mosquitto )。MQTT 代理扮演网关的角色，通过“mqtt”协议连接到许多不同类型的传感器。您的集群附带模拟脚本发布到的嵌入式 MQTT 代理。为方便起见，我们将使用 NiFi 来运行脚本而不是 Shell 命令。
Fayson在前面的文章介绍了什么是NiFi，参考《0622-什么是Apache NiFi》。同时对如何在CDH中使用Parcel安装CFM做了介绍，参考《0623-6.2.0-如何在CDH中安装CFM》。本文会首先对NiFi的使用做一下简单的介绍，然后对处理器（Processor）进行详细介绍。
Apache NiFi是一个强大的、可扩展的开源数据流处理工具，广泛应用于大数据领域。本文将介绍Apache NiFi的核心概念和架构，并提供代码实例展示其在实时数据流处理中的应用。
NiFi DataFlow Manager(DFM)用户可能会发现在单个服务器上使用一个NiFi实例不足以处理他们拥有的数据量。因此，一种解决方案是在多个NiFi服务器上运行相同的数据流。但是，这会产生管理问题，因为每次DFM想要更改或更新数据流时，他们必须在每个服务器上进行这些更改，然后单独监视每个服务器。通过集群NiFi服务器，可以增加处理能力以及单个接口，通过该接口可以更改数据流并监控数据流。集群允许DFM仅进行一次更改，然后将更改复制到集群的所有节点。通过单一接口，DFM还可以监视所有节点的健康状况和状态。
在过去的几周中，我进行了四个现场的NiFi演示会议，在不同地理区域有1000名与会者，向他们展示了如何使用NiFi连接器和处理器连接到各种系统。我要感谢大家参与和出席这些活动！如今，当在家中远程工作成为一种规范时，我们都需要交互式的演示会议和实时问答。如果您还没有看过我的现场演示会议，可以在这里观看，视频还没有过期。
本教程介绍了Apache Kafka的核心概念及其在可靠性、可伸缩性、持久性和性能至关重要的环境中所扮演的角色。
本教程涵盖了Apache NiFi的核心概念及其在其中流量管理，易用性，安全性，可扩展架构和灵活扩展模型非常重要的环境中所扮演的角色。
在组件工具栏下的NiFi屏幕顶部附近有一个条形,称为状态栏。它包含一些关于NiFi当前健康状况的重要统计数据：活动线程的数量可以指示NiFi当前的工作状态,排队统计数据表示当前在整个流程中排队的FlowFile数量以及这些FlowFiles的总大小。
DataFlow Manager(DFM)是NiFi用户，具有添加，删除和修改NiFi数据流组件的权限。
在本次实验中，您将实施一个数据管道来处理之前从边缘捕获的数据。您将使用 NiFi 将这些数据摄取到 Kafka，然后使用来自 Kafka 的数据并将其写入 Kudu 表。
Max Timer Driven Thread Count 和 Max Event Driven Thread Count
为了创建高效的数据流处理流程,需要了解可用的处理器（Processors ）类型，NiFi提供了大约近300个现成的处理器。这些处理器提供了可从不同系统中提取数据,路由,转换,处理,拆分和聚合数据以及将数据分发到多个系统的功能。如果还不能满足需求，还可以自定义处理器。
列名转换是ETL过程中常常遇到的场景。例如来源表user的主键id,要求写入目标表user的uid字段内,那么就需要列名转换.
NiFi的基本设计理念是基于数据流的编程Flow-Based Programming（FBP），应用是由处理器、连接器组成的网络。数据进入一个节点，由该节点对数据进行处理，根据不同的处理结果将数据路由到后续的其他节点进行处理。这是NiFi的流程比较容易可视化的一个原因。以下是NiFi的一些概念：
NiFi Connection是在两个已连接的NiFi处理器组件之间临时保存FlowFiles的位置。每个包含排队的NiFi FlowFiles的Connection在JVM堆中都会占一些空间。本文将对Connection进行分析，探究NiFi如何管理在该Connection中排队的FlowFiles和Connection对堆和性能的影响。
如下如.想象一下有 20 个这样的生成 UpdateAttribute 处理器,希望后续处理器分隔文本。现在，您需要将 SplitText 处理器替换为其他处理器。这样做将是一项困难的工作，因为它直接连接到 SplitText 处理器。但是，如果它们之间有一个漏斗，则只需替换漏斗的目标，而不是更换所有处理器
简单地说，NiFi就是为了实现系统间数据流的自动化而构建的。虽然术语“数据流”用于各种上下文，但我们在此处使用它来表示系统之间的自动和管理信息流。这个问题空间一直存在，因为企业有多个系统，其中一些系统创建数据，一些系统消耗数据。已经讨论并广泛阐述了出现的问题和解决方案模式。企业集成模式[eip]中提供了一个全面且易于使用的表单。
案例：使用NiFi将某个目录下产生的json类型的日志文件导入到Hive。这里首先将数据通过NiFi将Json数据解析属性，然后手动设置数据格式，将数据导入到HDFS中，Hive建立外表映射此路径实现外部数据导入到Hive中。
简介：本文主要讲解Apache NIFI的调度策略，对象主要是针对Processor组件。本文假定读者已经对Apache NIFI有了一定的了解和使用经验，同时作者也尽可能的去讲解的更透彻，使得本文尽可能让对NIFI接触不深的读者也能够看懂。
该处理器使用Hive流将流文件数据发送到Apache Hive表。传入的流文件需要是Avro格式，表必须存在于Hive中。有关Hive表的需求(格式、分区等)，请参阅Hive文档。分区值是根据处理器中指定的分区列的名称，然后从Avro记录中提取的。注意:如果为这个处理器配置了多个并发任务，那么一个线程在任何时候只能写入一个表。写入同一表的其他任务将等待当前任务完成对表的写入。
Cloudera Data Flow(CDF)作为Cloudera一个独立的产品单元，围绕着实时数据采集，实时数据处理和实时数据分析有多个不同的功能模块，如下图所示：
本文简单的讨论一下Apache NIFI项目结构的类资源隔离机制，适合接触过源码的同学阅读。
系统正在积极处理的FlowFiles保存在JVM内存中的Hash Map中。这使它们的处理效率非常高，但是由于多种原因，例如断电，内核崩溃，系统升级和维护周期，因此需要一种辅助机制来在整个进程重新启动中提供数据的持久性。FlowFile存储库是系统中当前存在的每个FlowFiles的元数据的Write-Ahead Log（或数据记录）。该FlowFile元数据包括与FlowFile相关联的所有attributes，指向FlowFile实际内容的指针（该内容存在于内容存储库中）以及FlowFile的状态，例如FlowFile所属的Connection/Queue。预写日志为NiFi提供了处理重启和意外系统故障所需的弹性。
注意：以上需要在NiFi集群中的每个节点上创建“/root/test/logdata”文件，“logdata”是文件，而非目录。
以上案例需要用到的处理器有：“CaptureChangeMySQL”、“RouteOnAttribute”、“EvaluateJsonPath”、“ReplaceText”、“PutHiveQL”。
处理器是最常用的组件,因为它负责数据的流入,流出,路由和操作，有许多不同类型的处理器，将处理器拖动到画布上时,会向用户显示一个对话框,以选择要使用的处理器类型。
NiFi对其摄取的每个数据保存明细。当数据通过系统处理并被转换,路由,拆分,聚合和分发到其他端点时,这些信息都存储在NiFi的Provenance Repository中。为了搜索和查看此信息,我们可以从全局菜单中选择数据源(Data Provenance)，也可以在对应的处理器上右键选择“View data provenance”进行查看。
本文通过Groovy，Jython，Javascript(Nashorn)和JRuby中的代码示例，介绍了有关如何使用Apache NiFi处理器ExecuteScript完成某些任务的各种方法。本文中的内容包括：
Apache NiFi是什么？NiFi官网给出如下解释：“一个易用、强大、可靠的数据处理与分发系统”。通俗的来说，即Apache NiFi 是一个易于使用、功能强大而且可靠的数据处理和分发系统，其为数据流设计，它支持高度可配置的指示图的数据路由、转换和系统中介逻辑。 为了对NiFi能够表述的更为清楚，下面通过NiFi的架构来做简要介绍，如下图所示。
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的核心组件。
在上一篇文章中，我们从安装在智能车辆上的传感器收集数据，并描述了ROS嵌入式应用程序，以准备用于训练机器学习（ML）模型的数据。本文展示了从边缘到云中数据湖的数据流。数据采用图像的形式以及与我们的自动驾驶汽车收集的每个图像相关的元数据（例如，IMU信息，转向角，位置）。我们将数据流定向到ClouderaDistribution Hadoop（CDH）集群，在该集群中将存储和整理数据以训练模型。
在最近的日常后台开发中经常遇到定时任务的需求，如定时通知、定时检查等重要的需求，绝对时间一定不会是完全准确的，它对于一个运行中的分布式系统其实没有太多指导意义，但是由于相对时间的计算不依赖于外部的系统，所以它的计算可以做的比较准确，这里简单总结一下定时任务在Go中的实现
Cloudera 在为流处理提供综合解决方案方面有着良好的记录。Cloudera 流处理 (CSP) 由 Apache Flink 和 Apache Kafka 提供支持，提供完整的流管理和有状态处理解决方案。在 CSP 中，Kafka 作为存储流媒体底层，Flink 作为核心流处理引擎，支持 SQL 和 REST 接口。CSP 允许开发人员、数据分析师和数据科学家构建混合流数据管道，其中时间是一个关键因素，例如欺诈检测、网络威胁分析、即时贷款批准等。
鼠标双击处理器或者选择以上“Configure”，打开配置处理器选项，配置分为四个部分：SETTINGS,SCHEDULING,PROPERTIES,COMMENTS。
在之前的官方文档Apache NiFi Overview一章我们有看到：对于任何基于组件的系统，涉及依赖的问题时常发生。NiFi通过提供自定义类加载器来解决这个问题，确保每个扩展包都暴露在一组非常有限的依赖中。因此，构建扩展包的时候不必担心它们是否可能与另一个扩展包冲突。这些扩展包的概念称为“NiFi Archives”，在Developer’s Guide中有更详细的讨论。
本文为用户使用Apache NiFi最新版本来监听SMTP邮件，并以编程方式做出反应以及捕捉数据提供了指导。 首先就可以注意到Apache NiFi 1.0.0应用了很棒的新界面，更加清晰也更加方便使
NIFI中文文档地址:https://nifichina.gitee.io/ 更新日志 2020-05-21 新增TailFile 新增ExecuteScript 新增探索 Apache NIFI 集群的高可用 2020-05-18 The 4 V’s of Big Data 2020-05-18 新增AttributeRollingWindow 新增CompareFuzzyHash 新增Apache NIFI入门(读完即入门) 新增了解NiFi最大线程池和处理器并发任务设置 新增深入理解NIFI Conn
随着越来越多的人呆在家里，让我们的信用卡代步，网上购物正在兴起。不幸的是，与这一趋势保持同步的是信用卡欺诈的增加。
面试者内心狂喜，这题刚背过：『Java内存主要分为五大块：堆、方法区、虚拟机栈、本地方法栈、PC寄存器，balabala……』
一些处理器支持配置运行持续时间(Run Duration)。此设置告诉处理器在单个任务中继续使用同一task尽可能多地来处理来自传入队列的的FlowFiles(或成批的流文件)。 对于处理单个任务本身非常快并且FlowFile数量也很大的处理器来说，这是一个理想的选择。
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频
