文 | 郑书国

NiFi之前是在美国国家安全局（NSA）开发和使用了8年的一个可视化、可定制的数据集成产品。2014年NSA将其贡献给了Apache开源社区，2015年7月成功成为Apache顶级项目。

Apache NiFi 是一个易于使用、功能强大而且可靠的数据处理和分发系统。Apache NiFi 是为数据流设计，它支持高度可配置的指示图的数据路由、转换和系统中介逻辑，支持从多种数据源动态拉取数据。简单地说，NiFi是为自动化系统之间的数据流而生。 这里的数据流表示系统之间的自动化和受管理的信息流。 基于WEB图形界面，通过拖拽、连接、配置完成基于流程的编程，实现数据采集、处理等功能。

下面是NIFI官网的一张图片，可以根据它了解一下NiFi的大体特征：

NiFi的基本设计理念是基于数据流的编程 Flow-Based Programming（FBP）。应用是由处理器黑盒、连接器组成的网络。数据进入一个节点，由该节点对数据进行处理，根据不同的处理结果将数据路由到后续的其他节点进行处理。这是NiFi的流程比较容易可视化的一个原因。

NiFi中的核心概念与FBP术语对比参照如下：

参照上述表格，简单来讲 FlowFile 是在各个节点间流动的数据；FlowFileProcessor 是数据的处理模块；Connection是各个处理模块间的一个队列；Flow Controllers是复杂流程的调度；Process Group封装流程的层次关系。

NiFi架构

NiFi是基于Java的，NiFi的核心部件在JVM里的位置如下图所示：

作为功能强大的数据处理和分发组件，NiFi自然原生支持集群部署方式（推荐部署方式）。NiFi集群部署模式如下图：

集群模式下，NiFi集群中的每个节点对数据执行相同的任务，但是每个节点都在不同的数据集上进行操作。和大部分大数据组件一样，NiFi集群使用Apache ZooKeeper提供协调服务。 Apache ZooKeeper选择一个NiFi节点作为集群协调器，故障转移由ZooKeeper自动处理。 所有集群节点向集群协调器报告心跳和状态信息。集群协调器负责节点的断开和连接。 此外，ZooKeeper会为每个集群选举一个节点作为集群主节点。 作为DataFlow管理器，您可以通过任何节点的用户界面（UI）与NiFi集群进行交互。您所做的任何更改都会同步到集群中的所有节点，从而允许多个入口点。

 NiFi特点

关键特性包括：

• 基于web的用户界面：

通过用户界面（UI）进行设计、控制和监视

高度可配置：

数据丢失容错和保证交付

低延迟和高吞吐量

动态优先级的队列

流可以在运行时修改

背压 Back Pressure

数据溯源：

从始至终跟踪数据流，NiFi对数据流路径上的每次变化都会保留日志记录，从而保证结果的可追溯，以及数据的恢复、重播、审计和评估。通过日志记录数据的历史细节。

易于扩展：

自定义数据处理器

支持快速开发和有效的测试

• 支持流模板（Flow Templates）

• 安全：

SSL、SSH、HTTPS加密内容，等等……

支持插入式（Pluggable）/多角色（Multi-Role）的安全机制

因为NiFi可以对来自多种数据源的流数据进行处理，Hortonworks认为HDF（Hortonworks Dataflow，基于NiFi）平台非常适合用于物联网（IoAT）的数据处理。NiFi并非只能用于物联网，实际上，它可以用于所有种类的实时数据处理，比如预测分析、欺诈检测、大数据注入、资源评估等等。NiFi项目自身提供了90个数据处理器（Data  Processors），这其中包括了数据的编码、加密、压缩、转换、从数据流创建Hadoop的序列文件、同AWS交互、发送消息到Kafka、从Twitter上获取消息，以及其它等等。你可以在拖放风格的可视化界面上来配置这些数据处理器，把它们链接到一起，并在它们之间使用背压机制来进行流控。NiFi还提供了内置的自动扩展、请求复制、负载均衡和故障切换机制。

下面用一个完整的编程实例来直观认识一下NiFi基于WEB拖拽方式的编程模型。如下是将获取到的Syslog事件信息推送到HBase的编程结果模型示意图，当编程设计人员通过NiFi WEB UI完成下面的类似数据流图的编程设计后，整个功能编码也已经完成了，也就是说无需开发人员进行任何代码的编写即可以实现业务需求。NiFi天生就是基于流程的编程，基于WEB图形界面，通过拖拽、连接、配置完成编程。

写在最后

NiFi（https://nifi.apache.org/）是Apache支持下基于可视化流程设计的数据分发平台，是大数据的搬运、提取、推送、转换、聚合、分发的开源软件工具，能够与Hadoop生态系统的大数据存储和各种文件、REST服务、SOAP服务、消息服务等联合使用，构成一体化的数据流服务。

任何一个大数据分析的系统，都需要一个强大的数据拉取组件、数据仓储系统、数据处理引擎、任务调度引擎和流程设计界面。Hadoop和Spark的重点是在数据的存储和任务调度，R的重点是数据分析引擎，而数据拉取组件和流程则是NiFi的主要强项。

#include <memory>
#include <unordered_map>

// 使用 std::unique_ptr 管理内存
std::unique_ptr<std::unordered_map<std::string, int>> my_hash_table =
    std::make_unique<std::unordered_map<std::string, int>>();

#include <hadoop/Config.hh>
#include <hadoop/MapReduce.hh>

class WordCountMapper : public hadoop::Mapper<hadoop::String, hadoop::String, hadoop::String, hadoop::Int> {
public:
  hadoop::Int map(const hadoop::String& key, const hadoop::String& value) override {
    // 分割文本并映射单词为键，值设为 1
    std::vector<std::string> words = split(value.str());
    for (const auto& word : words) {
      return hadoop::make_pair(hadoop::String(word), hadoop::Int(1));
    }
  }
};

class WordCountReducer : public hadoop::Reducer<hadoop::String, hadoop::Int, hadoop::String, hadoop::Int> {
public:
  hadoop::Int reduce(const hadoop::String& key, hadoop::Sequence<hadoop::Int>& values) override {
    // 汇总相同单词出现的次数
    int sum = 0;
    for (const auto& value : values) {
      sum += value.get();
    }
    return hadoop::make_pair(key, hadoop::Int(sum));
  }
};

int main(int argc, char** argv) {
  // 创建一个 MapReduce 作业
  hadoop::Job job;
  job.setJar("/path/to/wordcount.jar");

  // 设置 Mapper 和 Reducer
  job.setMapper<WordCountMapper>();
  job.setReducer<WordCountReducer>();

  // 运行作业
  int success = job.waitForCompletion();
  if (success) {
    std::cout << "MapReduce 作业成功运行。" << std::endl;
  } else {
    std::cerr << "MapReduce 作业失败。" << std::endl;
  }

  return 0;
}

// 创建 SparkContext，它是与 Spark 集群的连接
SparkContext spark;

// 从文件中加载文本数据
RDD<string> lines = spark.textFile("input.txt");

// 将文本行拆分为单词
RDD<string> words = lines.flatMap(
  [](string line) -> vector<string> {
    istringstream iss(line);
    vector<string> result;
    string word;
    while (iss >> word) {
      result.push_back(word);
    }
    return result;
  }
);

// 对单词进行计数
RDD<pair<string, int>> wordCounts = words.map(
  [](string word) -> pair<string, int> {
    return make_pair(word, 1);
  }
).reduceByKey(
  [](int a, int b) { return a + b; }
);

// 将结果保存到文件中
wordCounts.saveAsTextFile("output.txt");