大数据NiFi（十九）：实时Json日志数据导入到Hive

​实时Json日志数据导入到Hive

案例：使用NiFi将某个目录下产生的json类型的日志文件导入到Hive。这里首先将数据通过NiFi将Json数据解析属性，然后手动设置数据格式，将数据导入到HDFS中，Hive建立外表映射此路径实现外部数据导入到Hive中。

使用到的处理器有：“TailFile”、“EvaluateJsonPath”、“ReplaceText”、“PutHDFS”四个处理器。

一、配置“TailFile”处理器

“TailFile”处理器作用是"Tails"一个文件或文件列表，在文件写入文件时从文件中摄取数据。监控的文件为文本格式，当写入新行时会接收数据。如果要Tail的文件是定期"rolled over(滚动)"的（日志文件通常是这样），则可以使用可选的"Rolling Filename Pattern"从已滚动的文件中检索数据,NiFi未运行时产生的滚动文件在NiFi重启后仍会监控到。建议将运行计划设置为几秒，不使用默认0秒运行，否则此处理器将消耗大量资源。此处理器不支持监控压缩的文件。

关于“TailFile”处理器的“Properties”配置的说明如下：

配置步骤如下：

1、创建“TailFile”处理器

​

2、配置“PROPERTIES”

​

注意： 以上需要在NiFi集群中的每个节点上创建“/root/test/jsonfile”文件，“jsonfile”是文件，而非目录。

二、配置“EvaluateJsonPath”处理器

“EvaluateJsonPath”处理器根据FlowFile的内容计算一个或多个JsonPath表达式。根据处理器的配置，这些表达式的结果被赋值给FlowFile属性，或者被写入FlowFile本身的内容。

通过添加用户自定义的属性来输入Jsonpath，添加的属性的名称映射到输出流中的属性名称,属性的值必须是有效的JsonPath表达式(例如：$.name)。"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"。

关于“EvaluateJsonPath”处理器的“Properties”配置的说明如下：

示例说明：

• 提取流文件json内容，作为输出流的属性。（注意：当输出选择flowfile-attribute时，即使jsonpath匹配不到值，流文件也会路由到matched）

输入json如下：

​

输出结果如下：

​

• 提取流文件json内容，作为输出流的内容。（注意：当选择flowfile-content时，用户只能自定义添加一个属性；如果jsonPath匹配不到，会路由到unmatched）

​

输出流内容：

​

介绍完“EvaluateJsonPath”如何使用，下面来配置，配置步骤如下：

1、创建“EvaluateJsonPath”处理器

​

2、配置“PROPERTIES”

​

3、连接“TailFile”处理器和“EvaluateJsonPath”处理器

​

三、配置“ReplaceText”处理器

“ReplaceText”处理器会替换正则表达式匹配到的FlowFile中的内容，生成新的FlowFile内容。这里我们使用“ReplaceText”处理器将上个处理器“EvaluateJsonPath”处理后的每个FlowFile内容替换成自定义的内容，这里自定义内容都是从FlowFile的属性中获取的值，按照“\t”制表符隔开，方便后期存储到HDFS中映射Hive表。

配置步骤如下：

1、新建“ReplaceText”处理器，配置“PROPERTIES”处理器

​

2、连接“EvaluateJsonPath”处理器与“ReplaceText”处理器，同时设置

​

在“EvaluateJsonPath”中设置“failure”与“unmatched”的传递关系为自动终止：

​

四、​​​​​​​​​​​​​​配置“PutHDFS”处理器

这里创建“PutHDFS”处理器将上游处理的数据写入HDFS目录中。配置步骤如下：

1、创建“PutHDFS”处理器并配置

​

2、连接“ReplaceText”处理器与“PutHDFS”处理器并配置

​

配置“ReplaceText”处理器“failure”的FlowFile传递关系为自动终止：

​

配置“PutHDFS”处理器“failure”和“success”的FlowFile传递关系为自动终止：

​

五、运行测试

1、在Hive中创建外表personinfo在Hive中创建外表personinfo

CREATE TABLE personinfo(
id int,
name string,
age int
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
LOCATION '/mycluster/personinfo'

​

2、启动NiFi处理数据流程，处理数据

向任意NiFi节点/root/test/jsonfile文件中写入以下数据写入以下数据：

echo "{\"id\":1,\"name\":\"zhangsan\",\"age\":18}" >> /root/test/jsonfile
echo "{\"id\":2,\"name\":\"lisi\",\"age\":19}" >> /root/test/jsonfile
echo "{\"id\":3,\"name\":\"wangwu\",\"age\":20}" >> /root/test/jsonfile
echo "{\"id\":4,\"name\":\"maliu\",\"age\":21}" >> /root/test/jsonfile
echo "{\"id\":5,\"name\":\"tianqi\",\"age\":22}" >> /root/test/jsonfile