NiFi实现数据存储到数据库

一、NiFi环境搭建

1. Docker拉取NiFi镜像

使用 Docker 拉取 Apache NiFi 镜像。

docker pull apache/nifi:1.26.0

2. 创建自定义容器互联网络

自定义一个 Docker 网络,让两个容器能够互相通过容器名互联。

docker network create nifi-net

3. NiFi数据持久化处理

可以先启动一个临时容器，将原始 conf 拷到宿主机：

# 创建一个临时容器（不启动 NiFi）
docker create --name nifi-temp apache/nifi:1.26.0

# 拷贝 conf 到宿主机
docker cp nifi-temp:/opt/nifi/nifi-current/conf D:/Tools/Docker/nifi/

# 删除临时容器
docker rm nifi-temp

4. 启动NiFi容器

启动 NiFi 容器， 命名为 nifi。

官方 Docker 镜像支持直接用环境变量来覆盖端口配置：

• NIFI_WEB_HTTP_PORT：设置 HTTP 端口
• NIFI_WEB_HTTP_HOST：绑定地址
• 不要设置 NIFI_WEB_HTTPS_PORT／NIFI_WEB_HTTPS_HOST，NiFi 就不会启用 HTTPS。

docker run -d --name nifi -e TZ="Asia/Shanghai" --network nifi-net --privileged=true -e SINGLE_USER_CREDENTIALS_USERNAME=admin -e SINGLE_USER_CREDENTIALS_PASSWORD=nifi1357924680 -e NIFI_WEB_HTTP_HOST=0.0.0.0 -e NIFI_WEB_HTTP_PORT=8080 -e NIFI_REMOTE_INPUT_SOCKET_PORT=10000 -p 8080:8080 -p 10000:10000 -p 9090:9090 -p 9091:9091 -p 9092:9092 -p 9093:9093 -p 9094:9094 -p 9095:9095 -v D:/Tools/Docker/nifi/conf:/opt/nifi/nifi-current/conf -v D:/Tools/Docker/nifi/database_repository:/opt/nifi/nifi-current/database_repository -v D:/Tools/Docker/nifi/flowfile_repository:/opt/nifi/nifi-current/flowfile_repository -v D:/Tools/Docker/nifi/content_repository:/opt/nifi/nifi-current/content_repository -v D:/Tools/Docker/nifi/provenance_repository:/opt/nifi/nifi-current/provenance_repository -v D:/Tools/Docker/nifi/logs:/opt/nifi/nifi-current/logs apache/nifi:1.26.0

• 这里开放了9090-9095端口作为备用对外端口，需要到防火墙里添加入站和出战规则
• 将配置文件、数据文件、日志文件挂载到磁盘上，方便修改和查看
• 指定市区为上海，解决默认环境下八小时时差的问题

访问 http://localhost:8080/nifi 即可访问 NiFi 页面。

二、MySQL数据库环境搭建

1. Docker拉取MySQL镜像

使用 Docker 拉取 MySQL 官方镜像。

docker pull mysql:8.0

2. 启动MySQL容器

启动 MySQL 容器，并将数据和配置映射到宿主机目录。

docker run -d --name mysql --network nifi-net -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=admin -p 3306:3306 -v D:/Tools/Docker/MySQL/data:/var/lib/mysql -v D:/Tools/Docker/MySQL/conf:/etc/mysql/conf.d mysql:8.0

• MYSQL_ROOT_PASSWORD：root 用户的密码
• MYSQL_DATABASE：启动时自动创建的数据库
• MYSQL_USER/MYSQL_PASSWORD：创建额外用户（可选，仅在需要时设置）
• 容器名 mysql 会自动成为同网络下的 DNS 名称
• 开放宿主机 3306 端口，供外部（如 Django 或其他服务）连接
• 将宿主机 D:/Tools/Docker/MySQL/data 挂载为数据库数据目录，保证数据持久化
• 将宿主机 D:/Tools/Docker/MySQL/conf 挂载为配置目录，方便定制配置

启动成功后，在 Docker Desktop 中也能看到运行中的 MySQL 容器：

3. 下载MySQL的JDBC驱动

官网（https://dev.mysql.com/downloads/connector/j/）下载 Platform Independent 版本的驱动。

解压zip包，将jar包放到 drivers 目录下。

三、PostgreSQL数据库环境搭建

1. Docker拉取PostgreSQL镜像

使用 Docker 拉取 PostgreSQL 镜像。

docker pull postgres:15

2. 启动PostgreSQL容器

启动 PostgreSQL 容器。

docker run -d --name postgres --network nifi-net -e POSTGRES_USER=admin -e POSTGRES_PASSWORD=admin -e POSTGRES_DB=admin -e PGDATA=/var/lib/postgresql/data/pgdata -v D:/Tools/Docker/PostgreSQL/pgdata:/var/lib/postgresql/data/pgdata -v D:/Tools/Docker/PostgreSQL/conf:/etc/postgresql -p 5432:5432 postgres:15

• POSTGRES_USER/_PASSWORD/DB：请根据实际情况修改
• 容器名 postgres 会自动成为 DNS 名称
• 这里开放了5432端口作为备用对外端口
• 将数据文件挂载到磁盘上，方便修改和查看

启动成功后，Docker Desktop 也可以看到我们的第二个容器。

3. 下载PostgreSQL的JDBC驱动

在项目根目录（也就是你执行 docker run 的地方）新建一个 drivers 文件夹：

mkdir drivers

官网（https://jdbc.postgresql.org/download/）下载 PostgreSQL 驱动，放到 drivers 目录下。

四、NiFi连接数据库

1. 挂载JDBC驱动重启NiFi容器

挂载驱动重启NiFi容器：

docker run -d --name nifi -e TZ="Asia/Shanghai" --network nifi-net --privileged=true -e SINGLE_USER_CREDENTIALS_USERNAME=admin -e SINGLE_USER_CREDENTIALS_PASSWORD=nifi1357924680 -e NIFI_WEB_HTTP_HOST=0.0.0.0 -e NIFI_WEB_HTTP_PORT=8080 -e NIFI_REMOTE_INPUT_SOCKET_PORT=10000 -p 8080:8080 -p 10000:10000 -p 9090:9090 -p 9091:9091 -p 9092:9092 -p 9093:9093 -p 9094:9094 -p 9095:9095 -v D:/Tools/Docker/nifi/conf:/opt/nifi/nifi-current/conf -v D:/Tools/Docker/nifi/database_repository:/opt/nifi/nifi-current/database_repository -v D:/Tools/Docker/nifi/flowfile_repository:/opt/nifi/nifi-current/flowfile_repository -v D:/Tools/Docker/nifi/content_repository:/opt/nifi/nifi-current/content_repository -v D:/Tools/Docker/nifi/provenance_repository:/opt/nifi/nifi-current/provenance_repository -v D:/Tools/Docker/nifi/logs:/opt/nifi/nifi-current/logs -v D:/Tools/Docker/my-dev/drivers/postgresql-42.6.2.jar:/opt/nifi/nifi-current/lib/postgresql-42.6.2.jar -v D:/Tools/Docker/my-dev/drivers/mysql-connector-j-9.3.0.jar:/opt/nifi/nifi-current/lib/mysql-connector-j-9.3.0.jar apache/nifi:1.26.0

相较于上面的命令，这次增加了jar包的路径，如果jar的路径没有错误，会顺利启动成功。

2. 测试NiFi是否成功连接PostgreSQL

启动成功后，我们需要测试 NiFi 是否可以连接到 PostgreSQL。

确认 Controller Service 已启用

NiFi 首页，点击画布，左侧 NiFi-Flow 点击设置（齿轮按钮）。

在 Controller Service 里创建的 DBCPConnectionPool，填写必要的信息：

• Database Connection URL：jdbc:postgresql://postgres:5432/admin；
• Database Driver Class Name：com.mysql.cj.jdbc.Driver vs org.postgresql.Driver；
• Database User：PostgreSQL 用户名。
• Password：PostgreSQL 用户密码。

填写完成后，点击 Controller Service 列表最右侧的 Enable 按钮（闪电图标），在新窗口里将 Scope 选择为 Service and referencing components，然后点击 Enable 按钮。

下图为连接成功的情形：

使用 Processor 验证数据库连接

在画布空白处双击或右键 → 「Create Processor」→ 选择 ExecuteSQL → 点「Add」。
双击打开配置面板，设置如下，点击「Apply」。

PROPERTIES：

• DBCP Connection Pooling Service：选择你刚启用的 PostgreSQL 连接池（比如我刚新建的 DBCPConnectionPool）
• SQL select query：填 SELECT version();（PostgreSQL 会返回当前版本字符串）

RELATIONSHIPS：

• Automatically Terminate/Retry Relationships 的 failure 事件选择 terminate。

再从 ExecuteSQL 连出一个 LogAttribute 来查看结果。

右键 ExecuteSQL 点击运行，再右键 LogAttribute 点击运行。

有成功运行的记录之后，右键 ExecuteSQL 点击 View data provenance 查看调用记录。

每条记录的左侧感叹号按钮可以看到详细内容页，点击查看可以在线查看。

输出内容如下，成功输出了版本号，说明 NiFi 成功连接到了 PostgreSQL。

3. 测试NiFi是否成功连接MySQL

同理，我们再新建一个 DBCPConnectionPool，填写必要的信息：

• Database Connection URL：jdbc:mysql://mysql:3306/admin?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
• Database Driver Class Name：com.mysql.cj.jdbc.Driver
• Database User：root
• Password：root

这里注意

使用 Processor 验证数据库连接，参考NiFi连接PostgreSQL的步骤。

五、安装pgAdmin查看PostgreSQL数据

再启动一个 pgAdmin 容器，把它加入到同一个 nifi-net 网络，然后通过它来可视化管理 PostgreSQL。

1. Docker拉取pgAdmin镜像

使用 Docker 拉取 pgAdmin 镜像。

docker pull dpage/pgadmin4:latest

2. 启动PostgreSQL容器

启动 pgAdmin 容器。

docker run -d --name pgadmin --network nifi-net -e PGADMIN_DEFAULT_EMAIL=admin@admin.com -e PGADMIN_DEFAULT_PASSWORD=admin -p 5050:80 dpage/pgadmin4

3. pgAdmin连接PostgreSQL

我们需要在 pgAdmin 中添加你的 PostgreSQL 服务器。

浏览器访问：http://localhost:5050，输入你在上面设置的邮箱（admin@admin.com）和密码（admin）登录。

右击左侧 Servers → Create → Server

General 标签页：

• Name 填 NiFi-PostgreSQL（自定义随意起名）
  Connection 标签页：
• Host name/address：postgres
• Port：5432
• Maintenance database：admin（实际数据库名）
• Username：admin
• Password：admin

点击 Save，若连接信息填写正确，则进入仪表盘页面。

4. 在pgAdmin创建表

展开 Databases → 选中 admin → Schemas，右键 Tables → Create → Table。

填写表的相关配置：
General 标签页：

• Name 填 test_user
  Columns 标签页，点 “+” 添加列：
• id，类型 serial，勾选 PRIMARY KEY
• name，类型 text
• age，类型 Intager
• email，类型 text
  点击 Save。

六、安装phpMyAdmin查看MySQL数据

再启动一个 phpMyAdmin 容器，把它加入到同一个 nifi-net 网络，然后通过它来可视化管理 MySQL。

1. Docker拉取phpMyAdmin镜像

使用 Docker 拉取 phpMyAdmin 镜像。

docker pull phpmyadmin/phpmyadmin:latest

2. 启动phpMyAdmin容器

启动 phpMyAdmin 容器。

docker run -d --name phpmyadmin --network nifi-net -e PMA_HOST=mysql -e PMA_USER=root -e PMA_PASSWORD=root -p 9000:80 phpmyadmin/phpmyadmin:latest

3. phpMyAdmin连接MySQL

浏览器访问：http://localhost:9000，自动连接到我们的 MySQL。

七、NiFi完成数据插入

依次创建并连接 Processor： GenerateFlowFile -> EvaluateJsonPath -> ConvertJSONToSQL -> PutSQL

在 GenerateFlowFile 里编写测试数据源JSON：

{
    "data": [
        {
            "id": "122",
            "name": "cc",
            "age": "20",
            "email": "test@111.com"
        },
        {
            "id": "123",
            "name": "bb",
            "age": "39",
            "email": "test@123.com"
        }
    ]
}

ConvertJSONToSQL 里对选择数据库链接和操作类型，填写需要操作的数据表名。

PutSQL 同理设置数据库链接。检查每一个 Processor 是否 Invalid，一般无效的原因是 REALTIONSHIPS 未设置或设置错误。

Start 所有 Processor，一路向下直到执行完成。

打开 pgAdmin，查看测试数据是否已入库。