Apache nifi demo 实验

news2025/3/4 20:34:50

Apache nifi 是个数据流系统，可以通过配置自定义的流程来实现数据的转换。

比如可以配置一个流程，读取数据库里的数据，再转换，最后保存到本地文件。

这样可以来实现一些数据转换的操作，而不用特地编写程序来导入导出。

- 相关资料
国内下载镜像: https://mirrors.aliyun.com/apache/nifi/2.2.0/, 里面有不同的版本，这里下载标准版的nifi-2.2.0-bin.zip
官方文档: https://nifi.apache.org/
processor类型的文档:https://nifi.apache.org/components

本文包含以下内容:
- 启动运行
- 一些概念
- 一个读取csv格式数据，转换为json格式数据，写入到本地磁盘的demo

启动运行:
只需要解压zip,进入bin目录，打开命令行，执行执行nifi start即可启动，启动后密码会生成到logs目录的nifi-app.log文件里，搜索password可以看到生成的账号密码

D:\nifi-2.2.0-bin\nifi-2.2.0\bin>nifi start
JAVA_HOME=C:\Program Files\Java\jdk1.8.0_161
NIFI_HOME=D:\ljhlab\nifi-2.2.0-bin\nifi-2.2.0

日志文件里搜索，可以看到：
2025-03-02 22:17:53,849 INFO [main] o.a.n.a.s.u.SingleUserLoginIdentityProvider

Generated Username [8b418056-cf0a-46a3-b08f-a8ba25563eca]
Generated Password [9Va1ZkYgiXtqnOUytqj63ympwhZOjqZi]
这个就是生成的账号密码

异常处理，可能会遇到jdk版本问题，请下载对于版本的jdk
Error: A JNI error has occurred, please check your installation and try again
Exception in thread "main" java.lang.UnsupportedClassVersionError: org/apache/nifi/bootstrap/BootstrapProcess has been compiled by a more recent version of the Java Runtime (class file version 65.0), this version of the Java Runtime only recognizes class file versions up to 52.0
启动失败，可以看到是jdk版本要 65 ，需要jdk21的意思。
修改nifi-env.cmd,添加，或者直接在环境变量里设置JAVA_HOME的值
set JAVA_HOME="D:\env\openjdk-22.0.1_windows-x64_bin\jdk-22.0.1"

- 进入网页·
访问https://localhost:8443 进入网页ui界面

实验demo:

下面进行一个读取本地csv文件，然后导入到另外目录成json格式的实验，这里为了演示方便，csv数据写死在一个输入组件里。

数据准备:

title,author,score
你杀了谁,东野圭吾,7.1
黄仁勋：英伟达之芯,斯蒂芬·威特,7.3

流程简介：就是不断添加不同类型的processor，比如获取文件的,转换数据格式的，然后用线将不同组件链接起来，比如读取文件，链接到解析csv，链接的时候选择relationship，每个组件都会有relationship

实验流程:
- 添加一个组件作为输出数据源
- 添加一个转换数据格式的组件,ConvertRecord,转换csv成json格式
- 添加一个组件，将ConvertRecord都内容输出到本地文件

添加一个GenerateFLowFile的组件，用来生成一个FlowFile,FlowFile可以理解为一行数据,即这个组件会输出一行数据，这里生成一个csv数据

刚添加时有问题就会有黄色标，可以点击查看错误，来修正

每个组件都有properties来设置属性，这里点开来设置，这个组件输出的内容

- 再添加一个处理器，ConvertRecord，这个processor可以配置一个reader和writer，这里的话配置一个csv reader，用来解析第一个处理器传进来的数据。

点开properties，看到设置值里是空的，点击create new service创建一个配置。

然后将两个组件链接起来

可以看到还是有警告，是因为还有relationship没有设置，每个都要设置，

像failure就是失败时的，如果不需要进一步处理可以勾选terminate。success的等会会和链接到一个写入文件的组件里，这里测试也可以暂时设置为terminate先。
如果看到这种警告，就是刚刚添加的service还有有启动，打开properties，点击三个点，点go to service,把服务启动就行，在这里也可以进一步配置这些服务的属性