> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kasoftware.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Apache Kafka 端口设置

> 将知行之桥连接到 Apache Kafka 以进行数据集成和同步。

Apache Kafka 端口允许通过向 Apache Kafka 推送数据或从 Apache Kafka 拉取数据，将 Apache Kafka 集成到数据流中。

## 核心功能

* 具有生产和消费能力的实时消息流
* 多种身份验证方案，包括 SASL 和 Kerberos
* 基于主题的消息组织与消费者组管理
* 可配置的读取时长，用于控制等待消息到达的时间

按照以下步骤将知行之桥连接到 Apache Kafka。

## 前提条件

要与 Apache Kafka 建立连接，用户至少必须具备：

* 一个 Apache Kafka 环境，其中包含有效的 Apache Kafka 主机服务器和服务器管理器（broker，通常是 Zookeeper 或 KRaft）。

* 从 Apache Kafka 主机到包含目标主题的集群中所有 Apache Kafka 服务器和服务器管理器的网络连接。

* 在 Apache Kafka 中创建的主题，用于向服务器发送消息和从服务器接收消息。

* 从知行之桥到 Apache Kafka 服务器的清晰网络路径。

<Note>
  知行软件不提供 Apache Kafka 服务器或服务器管理器的创建或配置方面的设置协助或支持。有关 Apache Kafka 环境设置的信息，请参阅 [Apache Kafka](https://kafka.apache.org/41/) 文档。
</Note>

## 建立连接

要允许知行之桥使用 Apache Kafka 中的数据，必须先建立到 Apache Kafka 的连接。有两种方法可以建立此连接：

* 将 Apache Kafka 端口添加到工作流。然后，在 **设置** 选项卡中，单击连接下拉列表旁边的 **创建**。
* 打开知行之桥**系统设置** 页面，然后打开 **连接** 选项卡。点击 **添加**，选择 **Apache Kafka**，然后点击 **下一步**。

<Note>
  - 仅在第一次创建连接时才需要登录过程。
  - 与 Apache Kafka 的连接可以跨多个 Apache Kafka 端口重复使用。
</Note>

### 输入连接设置

打开新的连接对话框后，按照以下步骤操作：

1. 提供所需信息：

   * **名称** — 连接的静态名称。
   * **类型** — 始终设置为 Apache Kafka。
   * **身份验证方案** — 用于连接的授权方案。选项有 **Auto**、**None**、**Plain**、**Scram** 和 **Kerberos**。
   * **用户** —（除 **None** 之外的所有方案）用于登录的 Apache Kafka 用户名。
   * **密码** — 上面输入用户的密码。
   * **Bootstrap Servers** — 用于建立到 Apache Kafka 初始连接的主机/端口对。如果连接到 Confluent Cloud，可以在集群设置中找到此信息。

2. 可选择点击 **高级** 以打开高级连接设置的下拉菜单。在大多数情况下，不需要这些设置。

3. 点击 **测试连接** 以确保知行之桥可以使用提供的信息连接到 Apache Kafka。如果发生错误，请检查所有字段并重试。

4. 点击 **添加连接** 以完成连接。

5. 在端口配置窗格的 **连接** 下拉列表中，选择新建的连接。

6. 在 **Topic** 字段中，输入要作为目标的 Apache Kafka 主题。

7. 点击 **保存更改**。

<Note>
  **高级** 选项卡上还有特定于数据源的身份验证和配置选项。此文档并未描述所有这些选项，但你可以在知行软件网站的 [在线帮助文件](https://www.kasoftware.com/articles.html) 页面找到数据源的详细信息。
</Note>

## 选择操作

与 Apache Kafka 建立连接后，必须选择 Apache Kafka 端口应执行的操作。下表概述了每个操作及其在知行之桥工作流中的位置：

| 操作        | 描述                                   | 工作流中的位置 |
| --------- | ------------------------------------ | ------- |
| [生产](#生产) | 接受来自文件或其他端口的输入数据并将其发送到 Apache Kafka。 | 结束      |
| [消费](#消费) | 检查队列中的消息，并通过输出路径沿工作流向下发送获得的任何数据。     | 中间      |

### 生产

**生产** 操作将输入数据发送到 Apache Kafka。此数据可以来自其他端口，也可以来自你手动上传到 Apache Kafka 端口输入选项卡的文件。Apache Kafka 端口会将输入数据发送到你在 **配置** 部分的 **Topic** 字段中输入的主题。

#### 高级消息键和分区配置

Apache Kafka 端口在向 Kafka 主题生产消息时支持指定消息键和分区，从而在多分区环境中提供对消息分发和排序的细粒度控制。

##### 消息键配置

可以使用知行之桥消息头 `MessageKey` 来指定 Kafka 消息键，或者通过 **高级** 选项卡中[高级设置](#高级设置)部分的 **Key Header** 字段来告知知行之桥使用输入消息上的哪个标头来确定消息键。此键决定消息发送到哪个分区，并确保分区内的消息排序。

##### 分区配置

可以使用以下可选设置来控制哪个分区接收消息：

* **Partition Header:** 定义端口用于设置分区的消息头。默认标头名称为 `MessagePartition`，但可以使用包含所需分区的任何消息头来覆盖它。
* **Partition:** 在 **高级** 选项卡的[高级设置](#高级设置)部分提供分区编号。

<Note>
  - 如果同时存在分区标头和特定分区编号，则消息头优先。
  - 如果既没有指定消息键也没有指定分区，Kafka 将在目标主题的所有分区中分发消息。
</Note>

### 消费

**消费** 操作会针对你在 **配置** 部分的 **Topic** 字段中输入的主题检查 Apache Kafka 队列中的消息。你必须为此操作设置以下字段：

* **Consumer Group ID**：指定端口创建的消费者应该属于哪个组。
* **Read Duration**：端口等待消息到达的时间长度（以秒为单位）。无论收到的消息数量如何，端口都将等待整个持续时间。

通过 **消费** 端口处理的数据进入输出选项卡，并向下传输到知行之桥工作流的下一个步骤。

## 其他连接配置

### 高级选项卡

#### 高级设置

| 设置                   | 说明                                                                                   |
| -------------------- | ------------------------------------------------------------------------------------ |
| **Partition**        | 要向其发送消息的分区编号。如果同时提供了此项和 **Partition Header**，则以后者为准。有关此项及后续设置的详细信息，请参阅[分区配置](#分区配置)。 |
| **Partition Header** | 用于指定目标分区的知行之桥消息头名称。默认值为 `MessagePartition`。                                          |
| **Key Header**       | 用于指定消息键的知行之桥消息头名称。默认值为 `MessageKey`。详情请参阅[消息键配置](#消息键配置)。                            |
| **本地文件方案**           | 用于为端口输出的消息分配文件名的方案。可以在文件名中动态使用宏，以包含标识符和时间戳等信息。有关更多信息，请参阅[宏](#宏)。                     |
| **其他驱动程序设置**         | 仅在特定用例中使用的隐藏属性。                                                                      |

#### 消息

| 设置            | 说明                                                                                                                                         |
| ------------- | ------------------------------------------------------------------------------------------------------------------------------------------ |
| **保存到已发送文件夹** | 勾选此项可将端口处理的文件复制到该端口的 Sent 文件夹。                                                                                                             |
| **已发送文件夹方案**  | 指示端口根据所选间隔对 Sent 文件夹中的消息进行分组。例如，**每周** 选项指示端口每周创建一个新的子文件夹，并将该周的所有消息存储在该文件夹中。空白设置表示端口将所有消息直接保存到 Sent 文件夹中。对于处理大量消息的端口，使用子文件夹有助于保持消息有序并提升性能。 |

#### 日志

| 设置           | 说明                                                                                                                                             |
| ------------ | ---------------------------------------------------------------------------------------------------------------------------------------------- |
| **日志级别**     | 端口生成日志的详细程度。请求支持时，请将其设置为 **Debug**。                                                                                                            |
| **日志子文件夹方案** | 指示端口根据所选间隔对 Logs 文件夹中的文件进行分组。**每周** 选项（默认值）指示端口每周创建一个新的子文件夹，并将该周的所有日志存储在该文件夹中。将此设置留空表示端口将所有日志直接保存到 Logs 文件夹中。对于处理大量交易的端口，使用子文件夹有助于保持日志有序并提升性能。 |
| **日志消息**     | 勾选此项可使已处理文件的日志条目包含文件本身的副本。如果禁用此功能，可能无法从 **输入** 或 **输出** 选项卡下载文件副本。                                                                             |

#### 其他

*其他设置用于特定用例。*

| 设置       | 说明                                                                              |
| -------- | ------------------------------------------------------------------------------- |
| **其他设置** | 允许你以分号分隔的列表配置隐藏端口设置（例如 `setting1=value1;setting2=value2`）。正常的端口用例和功能不应需要使用这些设置。 |

### 自动化选项卡

#### 自动化设置

*与端口自动处理文件相关的设置。*

| 设置           | 说明                                                                         |
| ------------ | -------------------------------------------------------------------------- |
| **发送**       | 到达端口的文件是否自动发送。                                                             |
| **重试间隔**     | 发送失败后重试前的分钟数。                                                              |
| **最大尝试次数**   | 端口处理文件的最大次数。成功与否基于服务器的成功确认。如果设置为 0，端口将无限次重试文件。                             |
| **接收**       | 端口是否应自动查询数据源。                                                              |
| **间隔**       | 自动查询尝试之间的时间间隔。                                                             |
| **整点后的分钟数**  | 小时计划的分钟偏移量。仅当上方的间隔设置为 *每小时* 时适用。例如，如果该值设置为 5，自动化服务将在 1:05、2:05、3:05 等时间下载。 |
| **时间**       | 尝试发生的一天中的具体时间。仅当上方的间隔设置为 *每天*、*每周* 或 *每月* 时适用。                             |
| **日期**       | 尝试发生的日期。仅当上方的间隔设置为 *每周* 或 *每月* 时适用。                                        |
| **分钟**       | 尝试下载前等待的分钟数。仅当上方的间隔设置为 *分钟* 时适用。                                           |
| **Cron 表达式** | 一个五段字符串，表示用于确定尝试发生时间的 cron 表达式。仅当上方的间隔设置为 *高级* 时适用。                        |

#### 性能

*与端口资源分配相关的设置。*

| 设置          | 说明                                                                                                               |
| ----------- | ---------------------------------------------------------------------------------------------------------------- |
| **最大工作线程数** | 此端口处理文件时从线程池消耗的最大工作线程数。如果已设置，此值会覆盖[高级设置](./Advanced-Settings.html)页面中[性能设置](./Performance-Settings.html)部分的默认设置。 |
| **最大文件数**   | 分配给端口的每个线程发送的最大文件数。如果已设置，此值会覆盖[高级设置](./Advanced-Settings.html)页面中[性能设置](./Performance-Settings.html)部分的默认设置。     |

### 警报选项卡

*与配置警报和服务级别协议 (SLA) 相关的设置。*

#### 端口电子邮件设置

在执行 SLA 之前，需要设置用于通知的电子邮件警报。点击 **配置警报** 会打开一个新的浏览器窗口并转到[设置页面](./Settings-Page.html)，你可以在其中设置系统范围的警报。有关更多信息，请参阅[警报](./Settings-Page.html#alerts)。

#### 服务级别协议 (SLA) 设置

SLA 允许你配置工作流中的端口预期发送或接收的数量，并设置期望满足该数量的时间范围。当 SLA 未满足时，知行之桥会发送电子邮件提醒用户，并将 SLA 标记为 *At Risk*，这意味着如果 SLA 很快仍未满足，它将被标记为 *Violated*。这让用户有机会介入并确定 SLA 未满足的原因，并采取适当措施。如果在风险期结束时 SLA 仍未满足，则 SLA 会被标记为违反，并再次通知用户。

要定义 SLA，请点击 **添加预期数量条件**。

* 如果端口具有单独的发送和接收操作，请使用单选按钮指定 SLA 适用于哪个方向。
* 将 **Expect at least** 设置为你期望处理的最小交易数（数量），然后使用 **Every** 字段指定时间范围。
* 默认情况下，SLA 每天生效。要更改此设置，请取消选中 **Everyday**，然后选中所需的星期几。
* 使用 **And set status to 'At Risk'** 指定 SLA 应何时标记为存在风险。
* 默认情况下，只有当 SLA 违反时才发送通知。要更改此设置，请勾选 **Send an 'At Risk' notification**。

以下示例显示了一个 SLA 配置：该端口预计在周一至周五每天接收 1000 个文件。如果在时间段结束前 1 小时仍未收到 1000 个文件，则会发送风险通知。

<img src="https://mintcdn.com/qiao/ilnLy7gsSdFqbKmI/public/images/sla_defined.png?fit=max&auto=format&n=ilnLy7gsSdFqbKmI&q=85&s=33e5282b3f7970e3e1fef056825d7772" alt="SLA 配置示例" width="1408" height="1550" data-path="public/images/sla_defined.png" />

<Note>
  如有必要，可以关闭 SLA 警报。这在维护时段可能很有用。点击导航栏上的 **设置**，然后导航到 **警报 > 常规警报**。点击平板和铅笔图标进行编辑，并取消勾选 **SLA 警报** 设置。
</Note>

## 宏

在文件命名策略中使用宏可以提高组织效率，并有助于理解数据上下文。通过在文件名中加入宏，可以动态包含标识符、时间戳和标头信息等相关信息，为每个文件提供有价值的上下文。这有助于确保文件名反映对组织很重要的详细信息。

知行之桥支持以下宏，它们都使用以下语法：`%Macro%`。

| 宏                       | 说明                                                                                                               |
| ----------------------- | ---------------------------------------------------------------------------------------------------------------- |
| ConnectorID             | 计算为端口的 ConnectorID。                                                                                              |
| Ext                     | 计算为端口当前正在处理的文件的扩展名。                                                                                              |
| Filename                | 计算为端口当前正在处理的文件名（包含扩展名）。                                                                                          |
| FilenameNoExt           | 计算为端口当前正在处理的文件名（不含扩展名）。                                                                                          |
| MessageId               | 计算为端口输出消息的 MessageId。                                                                                            |
| RegexFilename:*pattern* | 将 RegEx 模式应用于端口当前正在处理的文件名。                                                                                       |
| Header:*headername*     | 计算为当前正在处理的消息上目标标头（`headername`）的值。                                                                               |
| LongDate                | 计算为系统当前日期时间的长格式（例如 Wednesday, January 24, 2024）。                                                                 |
| ShortDate               | 计算为 yyyy-MM-dd 格式的系统当前日期（例如 2024-01-24）。                                                                         |
| DateFormat:*format*     | 计算为指定格式（`format`）的系统当前日期时间。有关可用的日期时间格式，请参阅[日期格式示例](./Date-Formatters.html#date-formats-with-literal-characters)。 |
| Vault:*vaultitem*       | 计算为指定保管库项的值。                                                                                                     |

### 示例

某些宏（如 %Ext% 和 %ShortDate%）不需要参数，但其他宏需要参数。所有带参数的宏都使用以下语法：`%Macro:argument%`

以下是一些带参数宏的示例：

* %Header:headername%：其中 `headername` 是消息上的标头名称。
* %Header:mycustomheader% 解析为输入消息上设置的 `mycustomheader` 标头值。
* %Header:ponum% 解析为输入消息上设置的 `ponum` 标头值。
* %RegexFilename:pattern%：其中 `pattern` 是正则表达式模式。例如，`%RegexFilename:^([\\w][A-Za-z]+)%` 匹配并解析为文件名中的第一个单词，且不区分大小写（`test_file.xml` 解析为 `test`）。
* %Vault:vaultitem%：其中 `vaultitem` 是[保管库](./Global-Settings-Vault.html)中的项目名称。例如，`%Vault:companyname%` 解析为保管库中存储的 `companyname` 项的值。
* %DateFormat:format%：其中 `format` 是接受的日期格式（详情请参阅[日期格式示例](./Date-Formatters.html#sample-date-formats)）。例如，`%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%` 解析为文件上的日期和时间戳。

还可以创建更复杂的宏，如以下示例所示：

* 在一个文件名中组合多个宏：`%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%%EXT%`
* 在宏外包含文本：`MyFile_%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%`
* 在宏内包含文本：`%DateFormat:'DateProcessed-'yyyy-MM-dd_'TimeProcessed-'HH-mm-ss%`