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

将 Apache NiFi 连接到 ClickHouse

Community Maintained

Apache NiFi

是一款开源工作流管理软件,旨在实现软件系统间的数据流自动化。它支持创建 ETL 数据管道,并内置超过 300 个数据处理器。本分步教程将演示如何将 Apache NiFi 同时作为数据源和目标连接到 ClickHouse,并加载示例数据集。

准备连接信息​

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

参数说明
HOST 和 PORT通常，在使用 TLS 时端口为 8443，不使用 TLS 时端口为 8123。
DATABASE NAME默认提供一个名为 default 的数据库，请填写您要连接的目标数据库名称。
USERNAME 和 PASSWORD默认用户名为 default。请使用适合您使用场景的用户名。

您的 ClickHouse Cloud 服务的详细信息可以在 ClickHouse Cloud 控制台中查看。 选择某个服务并点击 Connect：

ClickHouse Cloud 服务 Connect 按钮

选择 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 可访问的文件夹中，并记录其绝对路径

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

  1. 要在 Apache NiFi 中配置 Controller Service，点击“齿轮”按钮打开 NiFi Flow Configuration 页面

    高亮显示齿轮按钮的 NiFi Flow Configuration 页面

  2. 选择 Controller Services 选项卡，并点击右上角的 + 按钮添加一个新的 Controller Service

    高亮显示添加按钮的 Controller Services 选项卡

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

    在 Controller Service 选择对话框中高亮显示 DBCPConnectionPool

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

    Controller Services 列表中显示无效的 DBCPConnectionPool 并高亮显示齿轮按钮

  5. 在 Properties 部分中，输入以下值

PropertyValueRemark
Database Connection URLjdbc:ch:https://HOSTNAME:8443/default?ssl=true根据实际情况替换连接 URL 中的 HOSTNAME
Database Driver Class Namecom.clickhouse.jdbc.ClickHouseDriver
Database Driver Location(s)/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jarClickHouse JDBC 驱动 JAR 文件的绝对路径
Database UserdefaultClickHouse 用户名
PasswordpasswordClickHouse 密码

  1. 在 Settings 部分，将该 Controller Service 的名称修改为 “ClickHouse JDBC”，以便于识别

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

  2. 点击“闪电”按钮，然后点击 “Enable” 按钮，激活 DBCPConnectionPool Controller Service

    在 Controller Services 列表中高亮显示闪电按钮

    启用 Controller Service 的确认对话框

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

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

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

  1. 添加一个 ExecuteSQL 处理器，并配置相应的上游和下游处理器

    NiFi 画布中显示包含 ExecuteSQL 处理器的工作流

  2. 在 ExecuteSQL 处理器的 "Properties" 部分中，输入以下值

    PropertyValueRemark
    Database Connection Pooling ServiceClickHouse JDBC选择为 ClickHouse 配置的 Controller Service
    SQL select querySELECT * FROM system.metrics在此输入你的查询

  3. 启动 ExecuteSQL 处理器

    ExecuteSQL 处理器配置界面，属性已填写完成

  4. 为确认查询已成功处理，检查输出队列中的任意一个 FlowFile

    队列列表对话框，显示可供检查的 FlowFile

  5. 将视图切换为 "formatted" 以查看输出 FlowFile 的结果

    FlowFile 内容查看器，以 formatted 视图显示查询结果

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

  1. 要在单次插入中写入多行数据,首先需要将多条记录合并为单条记录。可以使用 MergeRecord 处理器来完成此操作

  2. 在 MergeRecord 处理器的"Properties"部分,输入以下值

    PropertyValueRemark
    Record ReaderJSONTreeReader选择适当的记录读取器
    Record WriterJSONReadSetWriter选择适当的记录写入器
    Minimum Number of Records1000将此值更改为更大的数字,以便合并最小数量的行来形成单条记录。默认为 1 行
    Maximum Number of Records10000将此值更改为大于"Minimum Number of Records"的数字。默认为 1,000 行

  3. 要确认多条记录已合并为一条,请检查 MergeRecord 处理器的输入和输出。注意输出是多条输入记录组成的数组

    输入

    MergeRecord 处理器输入显示单条记录

    输出

    MergeRecord 处理器输出显示合并后的记录数组

  4. 在 PutDatabaseRecord 处理器的"Properties"部分,输入以下值

    PropertyValueRemark
    Record ReaderJSONTreeReader选择适当的记录读取器
    Database TypeGeneric保留默认值
    Statement TypeINSERT
    Database Connection Pooling ServiceClickHouse JDBC选择 ClickHouse 控制器服务
    Table Nametbl在此处输入您的表名
    Translate Field Namesfalse设置为"false",以便插入的字段名称必须与列名匹配
    Maximum Batch Size1000每次插入的最大行数。此值不应低于 MergeRecord 处理器中"Minimum Number of Records"的值

  5. 要确认每次插入包含多行数据,请检查表中的行数是否至少按 MergeRecord 中定义的"Minimum Number of Records"值递增

    查询结果显示目标表中的行数

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