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

文章目录

实时Json日志数据导入到Hive

一、配置“TailFile”处理器

1、创建“TailFile”处理器

2、配置“PROPERTIES”

二、配置“EvaluateJsonPath”处理器

1、创建“EvaluateJsonPath”处理器

2、配置“PROPERTIES”

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

三、配置“ReplaceText”处理器

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

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

四、配置“PutHDFS”处理器

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

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

五、运行测试

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

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

3、查看结果

六、配置“ConvertRecord”处理器

1、创建“ConvertRecord”处理器

2、配置“PROPERTIES”中“Record Reader”配置

3、配置“PROPERTIES”中“Record Writer”：

4、连接“TailFile”处理器和“ConvertRecord”处理器

5、连接“ConvertRecord”处理器与“PutHDFS”处理器

七、运行测试

实时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

3、查看结果

NiFi页面：

hive中结果：

问题：当我们一次性向某个NiFi节点的“/root/test/jsonfile”文件中写入数据时，这时“EvaluateJsonPath”一个FlowFile中会有多条json数据，当获取json属性时，只会获取第一条json对应的属性。当数据流向下游“ReplaceText”处理器时，由于设置每行替换成指定格式的行，这时会出现将本批次所有行数据都替换成了第一行的json格式数据。如下图：

当一次性向tail的文件输入多条数据，我们不希望全部json行内容替换成第一行json内容，那么可以将“TailFile”处理器处理的数据直接传递给“ConvertRecord”处理器，将数据由json格式转换成自定义文本格式数据，再传递到“PutHDFS”处理器即可，所以解决以上问题，我们这里复用之前的“TailFile”和“PutHDFS”处理器即可，下面只需要配置“ConvertRecord”处理器即可。

六、配置“ConvertRecord”处理器

“ConvertRecord”根据配置的“记录读取器”和“记录写出控制器”来将记录从一种数据格式转换为另一种数据格式。

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

配置步骤如下：

1、创建“ConvertRecord”处理器

2、配置“PROPERTIES”中“Record Reader”配置

点击以上之后，进入配置：

配置好以上之后，需要启动：

3、配置“PROPERTIES”中“Record Writer”：

点击以上之后进入配置：

4、连接“TailFile”处理器和“ConvertRecord”处理器

5、连接“ConvertRecord”处理器与“PutHDFS”处理器

同时设置“ConvertRecord”的处理“failure”关系为自动终止：

七、运行测试

删除HDFS中原有的“/personinfo”路径，启动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

查询Hive表personinfo数据：

• 标签：导入到,NiFi,文件,配置,flowfile,Json,处理器,root,EvaluateJsonPath
  From： https://blog.51cto.com/lansonli/6459230