3.Nifi典型案例

NIFI 典型案例 [toc]

课程目标

1、离线同步 Mysql 数据到 DFS

2、Json 内容转换为 Hive 支持的文本格式

3、实时同步 Mysql 数据到 Hive

4、Kafka 的使用

1.离线同步 Mysql 数据到 hdfs

大数据数据仓库系统中，经常需要进行数据同步操作，可以使用 nifi 来进行灵活的全流程操作。

准备工作：

1. 启动 Mysql 服务(5.7 版本)，在 Mysql 中运行\资料\mysql\nifi_test.sql中的 SQL 语句。
2. 启动 Hadoop 集群(与 NiFi 集群在同一个可访问的局域网网段)

1.1 处理器流程

QueryDatabaseTable ——> ConvertAvroToJSON ——> SplitJson ——> PutHDFS

QueryDatabaseTable 读取 Mysql 数据，ConvertAvroToJSON 将数据转换为可阅读的 Json 格式，再通过 SplitJson 进行切割获得单独的对象，PutHDFS 将所有对象写入 HDFS 中。

1.2 处理器说明

QueryDatabaseTable

描述

生成 SQL 选择查询，或使用提供的语句，并执行该语句以获取其指定的“最大值”列中的值大于先前看到的最大值的所有行。查询结果将转换为 Avro 格式。几种属性都支持表达式语言，但不允许传入连接。变量注册表可用于为包含表达式语言的任何属性提供值。如果需要利用流文件属性来执行这些查询，则可以将 GenerateTableFetch 和/或 ExecuteSQL 处理器用于此目的。使用流技术，因此支持任意大的结果集。使用标准调度方法，可以将该处理器调度为在计时器或 cron 表达式上运行。该处理器只能在主节点上运行。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示所有默认值，以及属性是否支持 NiFi 表达式语言。

ConvertAvroToJSON

描述

​ 将 Binary Avro 记录转换为 JSON 对象。该处理器提供了 Avro 字段到 JSON 字段的直接映射，因此，生成的 JSON 将具有与 Avro 文档相同的层次结构。请注意，Avro 模式信息将丢失，因为这不是从二进制 Avro 到 JSON 格式的 Avro 的转换。输出 JSON 编码为 UTF-8 编码。如果传入的 FlowFile 包含多个 Avro 记录的流，则生成的 FlowFile 将包含一个 JSON Array，其中包含所有 Avro 记录或 JSON 对象序列。如果传入的 FlowFile 不包含任何记录，则输出为空 JSON 对象。空/单个 Avro 记录 FlowFile 输入可以根据“包装单个记录”的要求选择包装在容器中。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值。

SplitJson

描述

该处理器使用 JsonPath 表达式指定需要的数组元素，将 JSON 数组分割为多个单独的流文件。每个生成的流文件都由指定数组的一个元素组成，并传输到关系“split”，原始文件传输到关系“original”。如果没有找到指定的 JsonPath，或者没有对数组元素求值，则将原始文件路由到“failure”，不会生成任何文件。

该处理器需要使用人员掌握 JsonPath 表达式语言。

属性配置

在下面的列表中，必需属性的名称以粗体显示。任何其他属性(不是粗体)都被认为是可选的，并且指出属性默认值（如果有默认值），以及属性是否支持表达式语言。

PutHDFS

描述

将 FlowFile 数据写入 Hadoop 分布式文件系统（HDFS）

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示所有默认值，以及属性是否支持 NiFi 表达式语言。

1.3 操作

1.3.1 创建组

1.3.2 创建 QueryDatabaseTable

1.3.3 创建并配置 Mysql 连接池

1.3.3.1 创建连接池

1.3.3.2 配置连接池

Database Connection URL = jdbc:mysql://192.168.52.6:3306/nifi_test?characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
Database Driver Class Name = com.mysql.jdbc.Driver
#此处的jar包需要提前上传到nifi服务器中
Database Driver Location(s) = /export/download/jars/mysql-connector-java-5.1.40.jar
Database User = root
Password = 123456

1.3.3.3 启动连接池

1.3.4 配置 QueryDatabaseTable

Custom Query = select id,name,mobile,email,son_json from user_info_nifi limit 15

1.3.5 创建配置 ConvertAvroToJSON

1.3.5.1 创建配置 ConvertAvroToJSON

1.3.5.2 连接

1.3.5.3 负载均衡消费数据

1.3.6 创建配置 SplitJson

1.3.6.1 SplitJson 配置

JsonPath Expression = $.*

1.3.6.2 连接

1.3.7 创建配置 PutHDFS

Hadoop Configuration Resources = /export/download/config/hdfs-site.xml,/export/download/config/core-site.xml
Directory = /user/hive/warehouse/nifi_test.db/user_info_nifi
Conflict Resolution Strategy = append

1.3.8 运行查看效果

1.3.8.1 启动 QueryDatabaseTable，并查看队列中数据

1.3.8.2 启动 ConvertAvroToJSON，并查看队列中数据

1.3.8.3 启动 SplitJson，并查看队列中数据

1.3.8.4 启动 PutHDFS，并查看处理器接收和输出的数据

1.3.8.5 查看 HDFS 数据

2. Json 内容转换为 Hive 支持的文本格式

在向 HDFS 同步数据的示例中，我们保存的文本内容是 Json 格式的，如图：

如果数据需要被 Hive 的外部表所使用，那么目前的 Json 数据格式是不满足要求的，我们如何将 Json 格式数据转换为 Hive 所需要的文本格式呢？

2.1 处理器流程

QueryDatabaseTable ——> ConvertAvroToJSON ——> SplitJson ——> EvaluateJsonPath ——> ReplaceText ——> PutHDFS

这里的重点是，增加了 EvaluateJsonPath 和 ReplaceText 处理器，EvaluateJsonPath 用来提取 json 中的属性，ReplaceText 用来替换掉 FlowFile 中的内容，以使内容符合 Hive 外部表所支持的文本格式。

1. 将 Json 数据中的属性值提取出来；
2. 转换为\t 分割字段；\n 分割行数据的格式。

2.2 处理器说明

EvaluateJsonPath

描述

该处理器根据流文件的内容计算一个或多个 JsonPath 表达式。这些表达式的结果被写入到 FlowFile 属性，或者写入到 FlowFile 本身的内容中，这取决于处理器的配置。通过添加用户自定义的属性来输入 jsonpath，添加的属性的名称映射到输出流中的属性名称(如果目标是 flowfile-attribute;否则，属性名将被忽略)。属性的值必须是有效的 JsonPath 表达式。“auto-detect”的返回类型将根据配置的目标进行确定。当“Destination”被设置为“flowfile-attribute”时，将使用“scalar”的返回类型。当“Destination”被设置为“flowfile-content”时，将使用“JSON”返回类型。如果 JsonPath 计算为 JSON 数组或 JSON 对象，并且返回类型设置为“scalar”，则流文件将不进行修改，并将路由到失败。如果所提供的 JsonPath 计算为指定的值，JSON 的返回类型可以返回“scalar”。如果目标是“flowfile-content”，并且 JsonPath 没有计算到一个已定义的路径，那么流文件将被路由到“unmatched”，无需修改其内容。如果目标是“flowfile-attribute”，而表达式不匹配任何内容，那么将使用空字符串创建属性作为值，并且 FlowFile 将始终被路由到“matched”。

属性配置

在下面的列表中，必需属性的名称以粗体显示。任何其他属性(不是粗体)都被认为是可选的，并且指出属性默认值（如果有默认值），以及属性是否支持表达式语言。

动态属性：

该处理器允许用户指定属性的名称和值。

应用场景

通常当需要从流文件 json 中提取某些数据作为流属性时，使用此处理器；或者从流文件 json 内容中提取一部分内容作为下一个流文件内容，使用此处理器。

ReplaceText

描述

使用其他值替换匹配正则表达式的流文件部分内容，从而更新流文件的内容。

属性配置

在下面的列表中，必需属性的名称以粗体显示。任何其他属性(不是粗体)都被认为是可选的，并且指出属性默认值（如果有默认值），以及属性是否支持表达式语言。

应用场景

使用正则表达式，来逐行或者全文本替换文件流内容，往往用于业务逻辑处理。

2.3 操作

2.3.1 EvaluateJsonPath 提取 Json 字段值

2.3.1.1 创建并连接 EvaluateJsonPath

2.3.1.2 将 Json 字段配置到 attribute

flowfile-attribute 即为将变量放置在属性中；

扩展属性就是我们读取到的 Json 属性。

同时处理把Invalid警告处理掉

2.3.1.3 启动查看结果

我们可以看到，经过 EvaluateJsonPath 处理后，FlowFile 的属性中，已经包含了 Json 的字段值。

2.3.2 ReplaceText 变更文本内容和格式

虽然我们已经获取到了 Json 中的具体字段值，但是可以看到，FlowFile 的内容还是 Json。如何替换掉 FlowFile 中的内容数据呢？

2.3.2.1 创建 ReplaceText 并连接

2.3.2.2 解决 Invalid

2.3.2.3 配置替换 FlowFile 内容

2.3.3 运行查看结果

2.3.3.1 运行所有处理器，查看最后的 FlowFile 输出

2.3.3.2 查看输出的 HDFS 文件

我们发现，最终的 HDFS 文件，已经符合 Hive 的格式要求，抽取完成。

3.实时同步 Mysql 数据到 Hive

3.1 处理器流程

​ NiFi 监控 MySQL binlog 处理调用流程如下: CaptureChangeMySQL ——> RouteOnAttribute ——> EvaluateJsonPath ——> ReplaceText ——> PutHiveQL

准备工作

Mysql 建库建表：

create table nifi_test.nifi_hive
(
    id       int auto_increment
        primary key,
    name     varchar(64) null,
    day_time date        null
);

Hive 建表：

CREATE TABLE myhive.nifi_hive(id int,name string,day_time string)
STORED AS ORC
TBLPROPERTIES('transactional'='true');

替换 Hive 支持 nar 包：

​ 上传文件NiFi\资料\nifi安装包\nifi-hive-nar-1.9.2.nar，将其替换到 NiFi 服务的 lib 目录下，并重启 NiFi 集群。

3.2 处理器说明

CaptureChangeMySQL

描述

从 MySQL 数据库检索更改数据捕获（CDC）事件。CDC 事件包括 INSERT，UPDATE，DELETE 操作。事件将作为单独的流文件输出，并按操作发生的时间排序。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值，属性是否支持NiFi 表达式语言以及属性是否被视为“敏感”，这意味着将加密其值。在敏感属性中输入值之前，请确保nifi.properties文件具有属性nifi.sensitive.props.key的条目。

写入属性

DistributedMapCacheServer

描述

提供可通过套接字访问的映射（键/值）缓存。与该服务的交互通常是通过 DistributedMapCacheClient 服务完成的。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值。

DistributedMapCacheClientService

描述

提供与 DistributedMapCacheServer 通信的功能。可以使用它来在 NiFi 群集中的节点之间共享地图

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值。

RouteOnAttribute

描述

该处理器使用属性表达式语言，根据流文件的属性去计算然后进行路由。该处理器往往用于判断逻辑。

属性配置

在下面的列表中，必需属性的名称以粗体显示。任何其他属性(不是粗体)都被认为是可选的，并且指出属性默认值（如果有默认值），以及属性是否支持表达式语言。

动态属性

该处理器允许用户指定属性的名称和值。

连接关系

自定义连接关系

可以根据用户配置处理器的方式创建动态连接关系。

PutHiveQL

描述

执行 HiveQL DDL / DML 命令（例如，UPDATE，INSERT）。预期传入 File 的内容是要执行的 HiveQL 命令。HiveQL 命令可以使用？转义参数。在这种情况下，要使用的参数必须作为 FlowFile 属性存在，命名约定为 hiveql.args.N.type 和 hiveql.args.N.value，其中 N 是一个正整数。hiveql.args.N.type 应该是指示 JDBC 类型的数字。FlowFile 的内容应采用 UTF-8 格式。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值。

HiveConnectionPool

描述

为 Apache Hive 提供数据库连接池服务。可以从池中请求连接，使用后返回连接。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示任何默认值，属性是否支持 NiFi 表达式语言。

3.3 操作

3.3.1 开启 Mysql 的 binlog 日志

Mysql 的版本号要求 5.7。

3.3.1.1 登陆 MySQL 查看日志状态

# mysql -u root -p123456
mysql> show variables like '%log_bin%';

3.3.1.2 退出 MySQL 登陆

mysql> exit

3.3.1.3 Linux 开启 binlog

编辑配置文件

vi /etc/my.cnf

行尾加上

server_id = 1
log_bin = mysql-bin
binlog_format = row

server-id :表示单个结点的 id,单个节点可以随意写，多个节点不能重复， log-bin 指定 binlog 日志文件的名字为 mysql-bin，以及其存储路径。

重启服务

systemctl restart mysqld.service

或者

service mysqld restart

重新登陆查询开启状态

3.3.1.4 Windows 开启 binlog

修改配置文件

找到 mysql 配置文件my.ini所在目录，一般在C:\ProgramData\MySQL\MySQL Server 5.7。

注意目录不是 C:\ Program Files \MySQL\MySQL Server 5.7。

server_id = 1
log_bin = mysql-bin
binlog_format = row

重启 Mysql 服务

查询开启状态

3.3.1.5 开启 Mysql 远程访问权限

GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '123456' WITH GRANT OPTION;
FLUSH PRIVILEGES;

3.3.2 实时获取 Mysql 变更

3.3.2.1 创建处理器组

​ 组名：MySqlToHive_Timely

3.3.2.2 创建 CaptureChangeMySQL

CaptureChangeMySQL 的配置中需要 DistributedMapCacheClientService、DistributedMapCacheServer 处理器，一并创建。

3.3.2.3 配置 DistributedMapCacheServer

3.3.2.4 配置 DistributedMapCacheClientService

3.3.2.5 启动 Cache 服务和客户端

​ 在模拟的集群模式下，因为三台服务都在同一台主机，所以会存在端口冲突的问题，但是并不影响使用。因为只要有一个节点的缓存服务启动正常就可以使用。

3.3.2.6 配置 CaptureChangeMySQL

MySQL Hosts = 192.168.52.6:3306
MySQL Driver Class Name = com.mysql.jdbc.Driver
MySQL Driver Location(s) = /export/download/jars/mysql-connector-java-5.1.40.jar
Username = root
Password = 123456
Include Begin/Commit Events = true
Include DDL Events = true

3.3.2.7 启动 CaptureChangeMysql

启动后报错：

FlowFile 数据的属性信息：

3.3.3 根据条件路由

3.3.3.1 RouteOnAttribute 多线程消费

根据自己的服务器硬件配置，以及数据的更新速率，进行评估后填写。

3.3.3.2 NiFi 表达式

NiFi 表达式官网：https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html

之前我们已经了解过 NiFi 表达式语言，这里我们仅针对 equals 函数进行说明。

NiFi 表达式的 equals 函数

equals

说明：equals函数使用非常广泛，它确定其主题是否等于另一个 String 值。请注意，该equals函数直接比较两个 String 值。注意不要将此函数与matchs函数混淆，后者会根据正则表达式评估其主题。

学科类型：任意

参数：

• value：用于比较 Subject 的值。必须与主题类型相同。

返回类型：布尔值

示例：我们可以使用表达式${filename:equals('hello.txt')}检查 FlowFile 的文件名是否为“ hello.txt” ，或者可以检查属性hello的值是否等于属性的值filename： ${hello:equals( ${filename} )}。

3.3.3.3 设置自定义属性

3.3.3.4 运行并查看输出

输出的数据内容：

{
  "type": "insert",
  "timestamp": 1582484253000,
  "binlog_filename": "mysql-bin.000005",
  "binlog_position": 375,
  "database": "nifi_test",
  "table_name": "nifi_hive_streaming",
  "table_id": 108,
  "columns": [
    {
      "id": 1,
      "name": "id",
      "column_type": 4,
      "value": 7
    },
    {
      "id": 2,
      "name": "name",
      "column_type": 12,
      "value": "testName5"
    },
    {
      "id": 3,
      "name": "day_time",
      "column_type": 91,
      "value": "2020-02-24"
    }
  ]
}

3.3.4 提取关键属性

​

​ EvaluateJsonPath 等处理器在提取数据时，可以使用 JsonPath 表达式，来灵活的获取信息。

3.3.4.1 JsonPath 表达式

简介

类似于 XPath 在 xml 文档中的定位，JsonPath 表达式通常是用来路径检索或设置 Json 的。

JsonPath 中的“根成员对象”始终称为$，无论是对象还是数组。

其表达式可以接受“dot–notation”和“bracket–notation”格式，例如

$.store.book[0].title、$[‘store’][‘book’][0][‘title’]

操作符

函数

可以在 JsonPath 表达式执行后进行调用，其输入值为表达式的结果。

过滤器

过滤器是用于过滤数组的逻辑表达式，一个通常的表达式形如：[?(@.age > 18)]，可以通过逻辑表达式&&或||组合多个过滤器表达式，例如[?(@.price < 10 && @.category == ‘fiction’)]，字符串必须用单引号或双引号包围，例如[?(@.color == ‘blue’)] or [?(@.color == “blue”)]。

示例

{
  "store": {
    "book": [
      {
        "category": "reference",
        "author": "Nigel Rees",
        "title": "Sayings of the Century",
        "price": 8.95
      },
      {
        "category": "fiction",
        "author": "Evelyn Waugh",
        "title": "Sword of Honour",
        "price": 12.99
      },
      {
        "category": "fiction",
        "author": "Herman Melville",
        "title": "Moby Dick",
        "isbn": "0-553-21311-3",
        "price": 8.99
      },
      {
        "category": "fiction",
        "author": "J. R. R. Tolkien",
        "title": "The Lord of the Rings",
        "isbn": "0-395-19395-8",
        "price": 22.99
      }
    ],
    "bicycle": {
      "color": "red",
      "price": 19.95
    }
  },
  "expensive": 10
}

3.3.4.2 提取 Json 属性到 Attribute

3.3.4.3 运行并查看输出

3.3.5 ReplaceText 转换 Sql

配置 ReplaceText

Replacement Value = insert into myhive.nifi_hive (id,name,day_time) values (${id},'${name}','${day_time}')

启动查看结果

3.3.6 写入 Hive

3.3.6.1 创建 PutHiveQL

略

3.3.6.2 创建配置 HiveConnectionPool

Database Connection URL = jdbc:hive2://192.168.52.120:10000
Hive Configuration Resources = /export/download/config/core-site.xml,/export/download/config/hdfs-site.xml,/export/download/config/hive-site.xml

配置完成后，记得启用 HiveConnectionPool。

3.3.6.3 PutHiveQL 关联 HiveConnectionPool

3.3.7 验证 Hive 表中是否成功写入数据

略。

4. Kafka 的使用

​ Kafka 是一个由 Scala 和 java 编写的高吞吐量的分布式发布订阅消息。它拥有很高的吞吐量、稳定性和扩容能力，在 OLTP 和 OLAP 中都会经常使用。使用 NiFi 可以简单快速的建立起 kafka 的生产者和消费者，而不需要编写繁杂的代码。

4.1 处理器说明

4.1.1 PublishKafka_0_10

描述

使用 Kafka 0.10.x Producer API 将 FlowFile 的内容作为消息发送到 Apache Kafka。要发送的消息可以是单独的 FlowFiles，也可以使用用户指定的定界符（例如换行符）进行定界。用于获取消息的辅助 NiFi 处理器是 ConsumeKafka_0_10。

属性配置

4.1.2 ConsumeKafka_0_10

描述

消耗来自专门针对 Kafka 0.10.x Consumer API 构建的 Apache Kafka 的消息。用于发送消息的辅助 NiFi 处理器是 PublishKafka_0_10。

属性配置

在下面的列表中，必需属性的名称以粗体显示。其他任何属性（非粗体）均视为可选。该表还指示所有默认值，以及属性是否支持 NiFi 表达式语言。

4.2 Producer 生产

4.2.1 创建处理器

​ 创建处理器组 kafka，进入组后分别创建 GenerateFlowFile 和 PublishKafka_0_10 处理器。

4.2.2 负载均衡生产消息

4.2.2.1 连接 GenerateFlowFile 和 PublishKafka_0_10

4.2.2.2 负载均衡并发

4.2.3 配置 GenerateFlowFile

4.2.3.1 调度配置：

每 1 秒生产一次数据

4.2.3.2 属性配置

文件大小 100b；每次生成 10 个相同文件；每次生成的流文件内容唯一。

4.2.4 配置 PublishKafka_0_10

4.2.4.1 属性配置

Brokers 设置为 192.168.52.100:9092,192.168.52.110:9092,192.168.52.120:9092

topic 设置为 nifi-topic，如果 topic 不存在，会自动创建；

Delivery Guarantee，对应 kafka 的 acks 机制，选择最为保险的 Guarantee Replicated Delivery，相当于 acks=all。

4.2.4.2 关系配置

4.2.5 启动流程并监听数据

4.2.5.1 启动流程

略

4.2.5.2 监听 kafka 消费数据

在 kafka 所在服务器执行监听命令：

/export/servers/kafka_2.11-0.10.2.1/bin/kafka-console-consumer.sh --bootstrap-server  192.168.52.110:9092 --topic nifi-topic

4.3 Consumer 消费

4.3.1 创建处理器并连接

创建 ConsumeKafka_0_10 和 LogAttribute 处理器，并连接。

4.3.2 配置 ConsumeKafka_0_10

Brokers 地址要和 Producer 的设置一样：192.168.52.100:9092,192.168.52.110:9092,192.168.52.120:9092

Topic 设置和 Producer 一致：nifi-topic

GroupId 随意设置：nifi

Offset Reset 设置为：latest，从最新的消息开始消费

4.3.3 设置 LogAttribute

设置为自连接

4.3.4 启动流程并查看日志

略、

4.3.5 增加生产频率

注意：如果服务器资源有限，不要进行此操作。

GenerateFlowFile 的调度频率加快：20ms