Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

将 Apache NiFi 连接到 ClickHouse

View as MarkdownEdit this pageReport an issueSet up docs MCP
由社区维护

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 管理员配置。

下载并运行 Apache NiFi

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

下载 ClickHouse JDBC 驱动

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

[object Object]

  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 jdbchttps://HOSTNAME:8443/default?ssl=truetrue 相应替换 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 密码

  1. 在 Settings 部分中，将 Controller Service 的名称改为“ClickHouse JDBC”，以便后续识别

    显示已填写属性的 DBCPConnectionPool 配置对话框

  2. 点击“lightning”按钮，然后点击“Enable”按钮，启用 DBCPConnectionPool Controller Service

    突出显示 lightning 按钮的 Controller Services 列表

    启用 Controller Service 的确认对话框

  3. 检查 Controller Services 选项卡，确认该 Controller Service 已启用

    显示已启用 ClickHouse JDBC 服务的 Controller Services 列表

[object Object]

  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 视图显示查询结果

[object Object]

  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 中！

Navigation