将 Apache NiFi 连接到 ClickHouse - ClickHouse Documentation

Apache NiFi 是一款开源工作流管理软件，用于自动化不同软件系统之间的数据流转。它支持创建 ETL 数据管道，并内置了 300 多种数据处理器。本分步教程将介绍如何将 Apache NiFi 连接到 ClickHouse，使其既可作为源端，也可作为目标端，并加载一个样本数据集。

准备连接信息

要通过 HTTP(S) 连接到 ClickHouse，你需要以下信息：

Parameter(s)	Description
`HOST` and `PORT`	通常，使用 TLS 时端口为 8443；不使用 TLS 时端口为 8123。
`DATABASE NAME`	默认情况下，存在一个名为 `default` 的数据库。请使用你要连接的数据库名称。
`USERNAME` and `PASSWORD`	默认情况下，用户名为 `default`。请根据你的使用场景使用相应的用户名。

你的 ClickHouse Cloud 服务的连接信息可在 ClickHouse Cloud 控制台中查看。选择一个服务，然后点击 Connect：选择 HTTPS。连接信息会显示在示例 curl 命令中。如果你使用的是自管理 ClickHouse，则连接信息由你的 ClickHouse 管理员配置。

下载并运行 Apache NiFi

对于全新部署，请从 https://nifi.apache.org/download.html 下载二进制程序，然后运行 ./bin/nifi.sh start 启动服务

下载 ClickHouse JDBC 驱动

访问 GitHub 上的 ClickHouse JDBC 驱动发布页面，查找最新的 JDBC 发布版本
在该发布版本页面中，点击 “Show all xx assets”，然后找到文件名中包含关键字 “shaded” 或 “all” 的 JAR 文件，例如 clickhouse-jdbc-0.5.0-all.jar
将该 JAR 文件放在 Apache NiFi 可访问的文件夹中，并记下其绝对路径

添加 `DBCPConnectionPool` Controller Service 并配置其属性

要在 Apache NiFi 中配置 Controller Service，请点击“齿轮”按钮，进入 NiFi Flow Configuration 页面
选择 Controller Services 选项卡，然后点击右上角的 + 按钮，添加新的 Controller Service
搜索 DBCPConnectionPool，然后点击“Add”按钮
新添加的 DBCPConnectionPool 默认处于 Invalid 状态。点击“齿轮”按钮开始配置
在“Properties”部分中，输入以下值

Property	值	备注
Database Connection URL	jdbc:ch:https://HOSTNAME:8443/default?ssl=true	将 connection URL 中的 HOSTNAME 替换为相应的主机名
Database Driver Class Name	com.clickhouse.jdbc.ClickHouseDriver
Database Driver Locations	/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jar	ClickHouse JDBC 驱动 JAR 文件的绝对路径
Database User	default	ClickHouse 用户名
Password	password	ClickHouse 密码

在 Settings 部分中，将 Controller Service 的名称改为“ClickHouse JDBC”，以便后续识别
点击“lightning”按钮，然后点击“Enable”按钮，启用 DBCPConnectionPool Controller Service
检查 Controller Services 选项卡，确认该 Controller Service 已启用

使用 `ExecuteSQL` 处理器从表中读取数据

添加 ExecuteSQL 处理器，以及相应的上游和下游处理器
在 ExecuteSQL 处理器的 “Properties” 部分中，填写以下值
Property Value Remark
Database Connection Pooling Service ClickHouse JDBC 选择为 ClickHouse 配置的 Controller Service
SQL select query SELECT * FROM system.metrics 在此输入查询
启动 ExecuteSQL 处理器
要确认查询已成功处理，请检查输出队列中的一个 FlowFile
将视图切换为 “formatted”，查看输出 FlowFile 的结果

Property	Value	Remark
Database Connection Pooling Service	ClickHouse JDBC	选择为 ClickHouse 配置的 Controller Service
SQL select query	SELECT * FROM system.metrics	在此输入查询

使用 `MergeRecord` 和 `PutDatabaseRecord` 处理器向表中写入数据

要在一次插入中写入多行，首先需要将多条记录合并为一条记录。这可以通过 MergeRecord 处理器实现

在 MergeRecord 处理器的 “Properties” 部分中，填入以下值

Property	Value	Remark
Record Reader	`JSONTreeReader`	选择合适的记录读取器
Record Writer	`JSONReadSetWriter`	选择合适的记录写入器
Minimum Number of Records	1000	将此值设大一些，以便至少合并这么多行来组成一条记录。默认值为 1 行
Maximum Number of Records	10000	将此值设为高于 “Minimum Number of Records” 的数字。默认值为 1,000 行

要确认多条记录已合并为一条记录，请检查 MergeRecord 处理器的输入和输出。注意，输出是一个包含多条输入记录的数组输入输出

在 PutDatabaseRecord 处理器的 “Properties” 部分中，填入以下值

Property	Value	Remark
Record Reader	`JSONTreeReader`	选择合适的记录读取器
Database Type	Generic	保持默认值
Statement Type	INSERT
Database Connection Pooling Service	ClickHouse JDBC	选择 ClickHouse 控制器服务
Table Name	tbl	在此输入你的表名
Translate Field Names	false	设为 “false”，这样插入的字段名必须与列名一致
Maximum Batch Size	1000	每次 insert 的最大行数。该值不应低于 `MergeRecord` 处理器中 “Minimum Number of Records” 的值

要确认每次 insert 都包含多行，请检查表中的行数是否每次至少按 MergeRecord 中定义的 “Minimum Number of Records” 的值递增。
恭喜——你已成功使用 Apache NiFi 将数据加载到 ClickHouse 中！

​准备连接信息

​下载并运行 Apache NiFi

​下载 ClickHouse JDBC 驱动

​添加 DBCPConnectionPool Controller Service 并配置其属性

​使用 ExecuteSQL 处理器从表中读取数据

​使用 MergeRecord 和 PutDatabaseRecord 处理器向表中写入数据