跳到主要内容
跳到主要内容

连接 Apache NiFi 和 ClickHouse

Community Maintained

Apache NiFi 是一个开源的工作流管理软件，旨在自动化软件系统之间的数据流。它允许创建 ETL 数据管道，并配备了超过 300 个数据处理器。这个逐步教程展示了如何将 Apache NiFi 连接到 ClickHouse，作为数据源和目的地，并加载示例数据集。

1. 收集你的连接详细信息​

To connect to ClickHouse with HTTP(S) you need this information:

  • The HOST and PORT: typically, the port is 8443 when using TLS or 8123 when not using TLS.

  • The DATABASE NAME: out of the box, there is a database named default, use the name of the database that you want to connect to.

  • The USERNAME and PASSWORD: out of the box, the username is default. Use the username appropriate for your use case.

The details for your ClickHouse Cloud service are available in the ClickHouse Cloud console. Select the service that you will connect to and click Connect:

Choose HTTPS, and the details are available in an example curl command.

If you are using self-managed ClickHouse, the connection details are set by your ClickHouse administrator.

ClickHouseにHTTP(S)で接続するには、次の情報が必要です：

  • HOSTとPORT: 通常、ポートはTLSを使用する場合は8443、TLSを使用しない場合は8123です。

  • DATABASE NAME: デフォルトでは、defaultという名前のデータベースがあります。接続したいデータベースの名前を使用してください。

  • USERNAMEとPASSWORD: デフォルトでは、ユーザー名はdefaultです。ご利用のケースに適したユーザー名を使用してください。

ClickHouse Cloudサービスの詳細はClickHouse Cloudコンソールで確認できます。接続するサービスを選択し、Connectをクリックします：

HTTPSを選択すると、詳細が例のcurlコマンドで提供されます。

セルフマネージドのClickHouseを使用している場合、接続の詳細はClickHouseの管理者によって設定されます。

2. 下载并运行 Apache NiFi​

  1. 对于新设置，从 https://nifi.apache.org/download.html 下载二进制文件，并通过运行 ./bin/nifi.sh start 启动

3. 下载 ClickHouse JDBC 驱动程序​

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

4. 添加 DBCPConnectionPool 控制器服务并配置其属性​

  1. 要在 Apache NiFi 中配置一个控制器服务，请通过点击“齿轮”按钮访问 NiFi 流配置页面

  2. 选择控制器服务选项卡，并通过点击右上角的 + 按钮添加一个新的控制器服务

  3. 搜索 DBCPConnectionPool 并点击“添加”按钮

  4. 新添加的 DBCPConnectionPool 默认处于无效状态。点击“齿轮”按钮开始配置

  5. 在“属性”部分，输入以下值

属性值备注
数据库连接 URLjdbc:ch:https://HOSTNAME:8443/default?ssl=true相应替换连接 URL 中的 HOSTNAME
数据库驱动类名com.clickhouse.jdbc.ClickHouseDriver
数据库驱动位置/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jarClickHouse JDBC 驱动程序 JAR 文件的绝对路径
数据库用户defaultClickHouse 用户名
密码passwordClickHouse 密码

  1. 在设置部分，将控制器服务的名称更改为“ClickHouse JDBC”，以便于引用

  2. 通过点击“闪电”按钮然后点击“启用”按钮来激活 DBCPConnectionPool 控制器服务


  3. 检查控制器服务选项卡，确保控制器服务已启用

5. 使用 ExecuteSQL 处理器从表中读取​

  1. 添加一个 ExecuteSQL 处理器，以及适当的上游和下游处理器

  2. 在 ExecuteSQL 处理器的“属性”部分，输入以下值

    属性值备注
    数据库连接池服务ClickHouse JDBC选择为 ClickHouse 配置的控制器服务
    SQL 查询SELECT * FROM system.metrics在这里输入你的查询

  3. 启动 ExecuteSQL 处理器

  4. 要确认查询是否已成功处理，检查输出队列中的一个 FlowFile

  5. 切换视图为“格式化”，查看输出 FlowFile 的结果

6. 使用 MergeRecord 和 PutDatabaseRecord 处理器写入表​

  1. 为了在单次插入中写入多行，我们首先需要将多个记录合并为一个记录。这可以通过 MergeRecord 处理器完成。

  2. 在 MergeRecord 处理器的“属性”部分，输入以下值

    属性值备注
    记录读取器JSONTreeReader选择适当的记录读取器
    记录写入器JSONReadSetWriter选择适当的记录写入器
    最小记录数1000将此值更改为更高的数字，以便合并的最小行数形成单个记录。默认为 1 行
    最大记录数10000将此值更改为高于“最小记录数”的数字。默认为 1000 行

  3. 要确认多个记录是否合并为一个，检查 MergeRecord 处理器的输入和输出。注意，输出是多个输入记录的数组

    输入

    输出

  4. 在 PutDatabaseRecord 处理器的“属性”部分，输入以下值

    属性值备注
    记录读取器JSONTreeReader选择适当的记录读取器
    数据库类型Generic保持默认
    语句类型INSERT
    数据库连接池服务ClickHouse JDBC选择 ClickHouse 控制器服务
    表名tbl在此输入你的表名
    转换字段名false设置为“false”，以便插入的字段名必须与列名匹配
    最大批量大小1000每次插入的最大行数。此值不应低于 MergeRecord 处理器中“最小记录数”的值

  5. 要确认每次插入包含多行，请检查表中的行计数是否至少增加了 MergeRecord 中定义的“最小记录数”值。

  6. 恭喜 - 你已成功使用 Apache NiFi 将数据加载到 ClickHouse 中！