> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. > 使用 NiFi 数据管道将数据流式写入 ClickHouse # 将 Apache NiFi 连接到 ClickHouse export const CommunityMaintainedBadge = () => { return
由社区维护
; }; export const Image = ({img, alt, size = "lg"}) => { const normalizedSize = ["sm", "md", "lg"].includes(size) ? size : "lg"; return
{alt}
; }; 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**：
ClickHouse Cloud 服务连接按钮
选择 **HTTPS**。连接信息会显示在示例 `curl` 命令中。
ClickHouse Cloud HTTPS 连接信息
如果你使用的是自管理 ClickHouse，则连接信息由你的 ClickHouse 管理员配置。 对于全新部署，请从 [https://nifi.apache.org/download.html](https://nifi.apache.org/download.html) 下载二进制程序，然后运行 `./bin/nifi.sh start` 启动服务 1. 访问 GitHub 上的 ClickHouse JDBC 驱动发布页面，查找最新的 JDBC 发布版本 2. 在该发布版本页面中，点击 "Show all xx assets"，然后找到文件名中包含关键字 "shaded" 或 "all" 的 JAR 文件，例如 `clickhouse-jdbc-0.5.0-all.jar` 3. 将该 JAR 文件放在 Apache NiFi 可访问的文件夹中，并记下其绝对路径 添加 DBCPConnectionPool Controller Service 并配置其属性} id="4-add-dbcpconnectionpool-controller-service-and-configure-its-properties"> 1. 要在 Apache NiFi 中配置 Controller Service，请点击“齿轮”按钮，进入 NiFi Flow Configuration 页面 突出显示齿轮按钮的 NiFi Flow Configuration 页面 2. 选择 Controller Services 选项卡，然后点击右上角的 `+` 按钮，添加新的 Controller Service 突出显示添加按钮的 Controller Services 选项卡 3. 搜索 `DBCPConnectionPool`，然后点击“Add”按钮 突出显示 DBCPConnectionPool 的 Controller Service 选择对话框 4. 新添加的 `DBCPConnectionPool` 默认处于 Invalid 状态。点击“齿轮”按钮开始配置 显示处于 Invalid 状态的 DBCPConnectionPool 且突出显示齿轮按钮的 Controller Services 列表 5. 在“Properties”部分中，输入以下值 | Property | 值 | 备注 | | -------------------------- | ---------------------------------------------------------------------------------------- | ------------------------------- | | Database Connection URL | jdbc[https://HOSTNAME:8443/default?ssl=true](https://HOSTNAME:8443/default?ssl=true)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 密码 | 6. 在 Settings 部分中，将 Controller Service 的名称改为“ClickHouse JDBC”，以便后续识别 显示已填写属性的 DBCPConnectionPool 配置对话框 7. 点击“lightning”按钮，然后点击“Enable”按钮，启用 `DBCPConnectionPool` Controller Service 突出显示 lightning 按钮的 Controller Services 列表
启用 Controller Service 的确认对话框 8. 检查 Controller Services 选项卡，确认该 Controller Service 已启用 显示已启用 ClickHouse JDBC 服务的 Controller Services 列表 使用 ExecuteSQL 处理器从表中读取数据} id="5-read-from-a-table-using-the-executesql-processor"> 1. 添加 `ExecuteSQL` 处理器，以及相应的上游和下游处理器 NiFi 画布，显示工作流中的 ExecuteSQL 处理器 2. 在 `ExecuteSQL` 处理器的 "Properties" 部分中，填写以下值 | Property | Value | Remark | | ----------------------------------- | ----------------------------- | ------------------------------------- | | Database Connection Pooling Service | ClickHouse JDBC | 选择为 ClickHouse 配置的 Controller Service | | SQL select query | SELECT \* FROM system.metrics | 在此输入查询 | 3. 启动 `ExecuteSQL` 处理器 已填写属性的 ExecuteSQL 处理器配置 4. 要确认查询已成功处理，请检查输出队列中的一个 `FlowFile` 队列列表对话框，显示可供检查的 FlowFile 5. 将视图切换为 "formatted"，查看输出 `FlowFile` 的结果 FlowFile 内容查看器，以 formatted 视图显示查询结果 使用 MergeRecord 和 PutDatabaseRecord 处理器向表中写入数据} id="6-write-to-a-table-using-mergerecord-and-putdatabaserecord-processor"> 1. 要在一次插入中写入多行，首先需要将多条记录合并为一条记录。这可以通过 `MergeRecord` 处理器实现 2. 在 `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 行 | 3. 要确认多条记录已合并为一条记录，请检查 `MergeRecord` 处理器的输入和输出。注意，输出是一个包含多条输入记录的数组 输入 显示单条记录的 MergeRecord 处理器输入 输出 显示合并后记录数组的 MergeRecord 处理器输出 4. 在 `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" 的值 | 5. 要确认每次 insert 都包含多行，请检查表中的行数是否每次至少按 `MergeRecord` 中定义的 "Minimum Number of Records" 的值递增。 显示目标端表中行数的查询结果 6. 恭喜——你已成功使用 Apache NiFi 将数据加载到 ClickHouse 中！