配置 Kafka 使用者时，可以配置使用者组名称、主题和 ZooKeeper 连接信息。

您可以将 Kafka 使用者配置为使用融合架构注册表。融合模式注册表是 Avro 模式的分布式存储层，它使用 Kafka 作为其底层存储机制。

您可以根据需要添加其他 Kafka 配置属性。您还可以将源配置为使用 Kafka 安全功能。

Kafka 使用者包括记录标头属性，使您能够在管道处理中使用有关记录的信息。

使用者组标识的 Kafka 使用者源首次收到来自主题的消息时，将为该使用者组和主题创建偏移条目。偏移条目是在动物园守护者或卡夫卡中创建的，具体取决于您的卡夫卡版本和代理配置。

您可以将自定义卡夫卡配置属性添加到卡夫卡使用者。

添加 Kafka 配置属性时，请输入确切的属性名称和值。Kafka 使用者不会验证属性名称或值。

Kafka 使用者源创建记录标头属性，其中包含有关记录的原始文件的信息。当源处理 Avro 数据时，它将 Avro 架构包含在 avroSchema 记录标头属性中。

您可以使用记录：属性或记录：属性或默认值函数来访问属性中的信息。有关使用记录标头属性的详细信息，请参阅使用标头属性。

您可以将卡夫卡消费者源配置为通过 SSL/TLS 和/或 Kerberos 进行安全连接。

执行以下步骤，使卡夫卡消费者源能够使用 SSL/TLS 连接到卡夫卡。您可以使用相同的步骤来配置 Kafka 创建器。

例如，以下属性允许阶段使用 SSL/TLS 通过客户端身份验证连接到卡夫卡：

使用 Kerberos 身份验证时，数据收集器将使用 Kerberos 主体和密钥选项卡连接到 Kafka。

执行以下步骤，使卡夫卡消费者源能够使用 Kerberos 连接到卡夫卡：

例如，以下卡夫卡属性允许使用 Kerberos 连接到卡夫卡：

您可以启用卡夫卡消费者源以使用 SSL/TLS 和 Kerberos 连接到卡夫卡。

Kafka 消费者来源根据数据格式以不同的方式处理数据。卡夫卡消费者可以处理以下类型的数据：

阿夫罗

为每条消息生成一条记录。包括每个十进制字段的 和 字段属性。precisionscale

源在 记录标头属性中包含 Avro 架构。您可以使用下列方法之一来指定 Avro 架构定义的位置： avroSchema

• 消息/数据包括架构 - 在消息中使用架构。
• 在管道配置中 - 使用在阶段配置中提供的架构。
• 融合架构注册表 - 从融合架构注册表中检索架构。融合模式注册表是 Avro 模式的分布式存储层。您可以将源配置为通过消息中嵌入的架构 ID 或阶段配置中指定的架构 ID 或主题在 Confluent 架构注册表中查找架构。

  您必须指定源用于反序列化消息的方法。如果 Avro 架构 ID 嵌入在每条消息中，请在 Kafka 选项卡上将键和值反序列化程序设置为“融合”。

在阶段配置中使用架构或从 Confluent 架构注册表中检索架构会覆盖消息中可能包含的任何架构，并且可以提高性能。

二元的

生成一条记录，该记录在记录的根目录中有一个字节数组字段。

当数据超过用户定义的最大数据大小时，源无法处理数据。由于未创建记录，因此源无法将记录传递到要作为错误记录写入的管道。相反，源会生成阶段错误。

数据报

为每条消息生成一条记录。源可以处理 收集到的消息、NetFlow 5 和 NetFlow 9 消息以及以下类型的系统日志消息：

• 目录 5424
• 目录 3164
• 非标准常见消息，例如没有版本数字的 RFC 3339 日期

处理网络流消息时，该阶段会根据网络流版本生成不同的记录。处理网络流 9 时，将根据网络流 9 配置属性生成记录。有关详细信息，请参阅 NetFlow 数据处理。

分隔

为每个分隔线生成一条记录。可以使用以下带分隔符的格式类型：

• 默认 CSV - 包含逗号分隔值的文件。忽略文件中的空行。
• RFC4180 CSV - 严格遵循 RFC4180 准则的逗号分隔文件。
• MS Excel CSV - Microsoft Excel 逗号分隔的文件。
• MySQL CSV - 以逗号分隔的 MySQL 文件。
• 制表符分隔值 - 包含制表符分隔值的文件。
• PostgreSQL CSV - PostgreSQL 逗号分隔的文件。
• PostgreSQL Text - PostgreSQL 文本文件。
• 自定义 - 使用用户定义的分隔符、转义符和引号字符的文件。
• 多字符分隔 - 使用多个用户定义的字符来分隔字段和行，以及单个用户定义的转义字符和引号字符的文件。

您可以对分隔数据使用列表或列表映射根字段类型，并可以选择包括标题行中的字段名称（如果可用）。有关根字段类型的详细信息，请参阅分隔数据根字段类型。

使用标题行时，可以启用处理具有其他列的记录。其他列使用自定义前缀和按顺序递增的整数（如_extra_1、_extra_2）命名。如果不允许其他列，则包含其他列的记录将发送到错误。

还可以将字符串常量替换为空值。

当记录超过为阶段定义的最大记录长度时，阶段将根据为该阶段配置的错误处理来处理对象。

断续器

为每个 JSON 对象生成一条记录。您可以处理包含多个 JSON 对象或单个 JSON 数组的 JSON 文件。

当对象超过为源定义的最大对象长度时，源将根据为阶段配置的错误处理来处理对象。

日志

为每个日志行生成一条记录。

当一条线超过用户定义的最大线长时，原点将截断较长的线。

您可以将已处理的日志行作为字段包含在记录中。如果日志行被截断，并且您在记录中请求日志行，则原点包括截断的行。

您可以定义要读取的日志格式或类型。

普罗托布夫

为每个原始消息生成一条记录。默认情况下，源假定消息包含多个原始消息。

Protobuf 消息必须与指定的消息类型匹配，并在描述符文件中进行描述。

当记录的数据超过 1 MB 时，源无法继续处理消息中的数据。源根据阶段错误处理属性处理消息，并继续读取下一条消息。

有关生成描述符文件的信息，请参阅 Protobuf 数据格式先决条件。

断续器记录

为每条记录生成一条记录。用于处理数据收集器管道使用 SDC 记录数据格式生成的记录。

对于错误记录，源提供从原始管线中的源读取的原始记录，以及可用于更正记录的错误信息。

处理错误记录时，源需要原始管道生成的错误文件名和内容。

发短信

基于自定义分隔符为每行文本或每段文本生成记录。

当直线或截面超过为原点定义的最大线长时，原点会截断原点。原点添加一个名为“截断”的布尔字段，以指示该行是否被截断。

有关使用自定义分隔符处理文本的详细信息，请参阅使用自定义分隔符的文本数据格式。

.XML

基于用户定义的分隔符元素生成记录。使用根元素正下方的 XML 元素或定义简化的 XPath 表达式。如果未定义分隔符元素，则源将 XML 文件视为单个记录。

默认情况下，生成的记录包括 XML 属性和命名空间声明作为记录中的字段。您可以配置阶段以将其作为字段属性包含在记录中。

您可以在字段属性中包含每个已分析的 XML 元素和 XML 属性的 XPath 信息。这还会将每个命名空间放在 xmlns 记录标头属性中。

当记录超过用户定义的最大记录长度时，原点将跳过该记录并继续处理下一条记录。它将跳过的记录发送到管道以进行错误处理。

使用 XML 数据格式处理有效的 XML 文档。有关 XML 处理的更多信息，请参见读取和处理 XML 数据。

配置卡夫卡使用者源以从卡夫卡群集读取消息。

配置卡夫卡使用者时，将配置常规属性，包括卡夫卡和动物园守护者详细信息。根据需要配置其他数据格式属性。您可以选择添加自定义卡夫卡属性。

1. 在“属性”面板的“常规”选项卡上，配置以下属性：

   一般财产	描述: __________
   名字	艺名。
   描述: __________	可选说明。
   舞台库	要使用的库版本。
   记录错误	阶段的错误记录处理： 丢弃 - 丢弃记录。 发送到错误 - 将记录发送到管道以进行错误处理。 停止管道 - 停止管道。对群集管道无效。

2. 在“卡夫卡”选项卡上，配置以下属性：

   卡夫卡 房产	描述: __________
   代理 URI	卡夫卡代理的连接字符串。使用以下格式：。<host>:<port> 要确保连接，请输入以逗号分隔的其他代理 URI 列表。
   动物园管理员乌里	卡夫卡群集的动物园管理员的连接字符串。使用以下格式：。<host>:<port> 要使用动物园管理员仲裁，请输入逗号分隔的列表。 要使用动物园管理员 chroot 路径，请在列表末尾添加该路径，如下所示： <span style="color:#333333"><code><host>:<port>, <host2>:<port2>, .../<chroot_path></code></span>
   消费者群体	数据收集器所属的 Kafka 使用者组。
   主题	卡夫卡主题阅读。
   制作单张唱片	对于每个分区，为包含多个对象的记录生成单个记录。 如果未选定该属性，则当一条记录包含多个对象时，源将生成多条记录。
   最大批大小（记录）	一次处理的最大记录数。接受的值最高可达数据收集器最大批大小。 默认值为 1000。数据收集器默认值为 1000。
   批量等待时间（毫秒）	发送部分或空批处理之前要等待的毫秒数。
   每个分区的速率限制（卡夫卡消息）	每个分区的每个批处理要读取的最大消息数。此属性仅在群集模式下使用，有助于近似 Kafka 源的有效批大小。
   自动偏移复位	当使用者组和主题的组合不存在偏移量时，确定要读取的第一条消息的方法： 最早 - 读取从主题中的第一条消息开始的消息。 最新 - 读取从主题中最后一条消息开始的消息。 无 - 如果不存在偏移，则生成错误。 时间戳 - 读取以“自动偏移重置时间戳”属性中指定的时间戳开头的消息。 您必须使用 Kafka 版本 0.10.1.0 或更高版本才能根据时间戳读取消息。 默认值为“最早”。
   自动偏移复位时间戳（毫秒）	当原点不存在偏移量时读取的最早消息的时间戳。以毫秒为单位指定自 epoch （1970 年 1 月 1 日） 以来的指定。使用时间戳方法确定要读取的第一条消息时可用。
   卡夫卡配置	要使用的其他卡夫卡配置属性。使用简单或批量编辑模式，单击添加图标以添加属性。定义卡夫卡属性名称和值。 按照 Kafka 的预期使用属性名称和值。 有关启用与 Kafka 的安全连接的信息，请参阅启用安全性。
   包括时间戳	在记录标头中包含卡夫卡时间戳。源从卡夫卡检索时间戳。如果 Kafka 消息没有时间戳，则源将记录标头中的时间戳属性保留为空。当管道写入多个 Kafka 集群时，启用此属性以保持时间戳在集群之间的一致性。 您必须使用 Kafka 版本 0.10 或更高版本才能在记录标头中包含 Kafka 时间戳。

3. 在“数据格式”选项卡上，配置以下属性：

   数据格式属性	描述: __________
   数据格式	要读取的数据类型。使用以下选项之一： 阿夫罗 二元的 数据报 分隔 断续器 日志 普罗托布夫 断续器记录 发短信 .XML

4. 对于 Avro 数据，在“数据格式”选项卡上，配置以下属性：

   阿夫罗 房产	描述: __________
   阿夫罗架构位置	处理数据时要使用的 Avro 架构定义的位置： 消息/数据包括架构 - 在消息中使用架构。 在“管道配置”中 - 使用阶段配置中提供的架构。 汇合架构注册表 - 从融合架构注册表中检索架构。 在阶段配置或融合架构注册表中使用架构可以提高性能。
   阿夫罗架构	用于处理数据的 Avro 架构定义。重写与数据关联的任何现有架构定义。 您可以选择使用 runtime：loadResource 函数来使用存储在运行时资源文件中的模式定义。
   架构注册表网址	用于查找架构的融合架构注册表 URL。若要添加 URL，请单击“添加”。使用以下格式输入 URL： <span style="color:#333333"><code>http://<host name>:<port number></code></span>
   查找架构依据	用于在融合模式注册表中查找模式的方法： 主题 - 查找指定的 Avro 架构主题。 架构 ID - 查找指定的 Avro 架构 ID。 嵌入式架构 ID - 查找每条消息中嵌入的 Avro 架构 ID。 覆盖与消息关联的任何现有架构定义。
   架构主题	Avro 架构主题在融合架构注册表中查找。 如果指定的使用者具有多个架构版本，则源将使用该主题的最新架构版本。若要使用较旧的版本，请查找相应的架构 ID，然后将“查找架构依据”属性设置为“架构 ID”。
   架构标识	用于在融合架构注册表中查找的 Avro 架构 ID。

5. 对于二进制数据，请在“数据格式”选项卡上配置以下属性：

   二进制属性	描述: __________
   最大数据大小（字节）	消息中的最大字节数。无法处理较大的邮件或将其写入错误。

6. 对于数据报数据，在“数据格式”选项卡上，配置以下属性：

   数据报属性	描述: __________
   数据报数据包格式	数据包格式的数据： 已收集 净流 系统日志 原始/分离数据
   类型DB 文件路径	用户提供的类型.db文件的路径。覆盖默认类型.db文件。 仅用于收集的数据。
   转换高分辨率时间和间隔	将收集的高分辨率时间格式间隔和时间戳转换为 UNIX 时间（以毫秒为单位）。 仅用于收集的数据。
   排除间隔	从输出记录中排除间隔字段。 仅用于收集的数据。
   身份验证文件	可选身份验证文件的路径。使用身份验证文件接受签名和加密的数据。 仅用于收集的数据。
   记录生成模式	确定要包含在记录中的值的类型。选择以下选项之一： 仅原始 仅解释 原始和解释 仅适用于网流 9 数据。
   缓存中的最大模板数	要在模板缓存中存储的最大模板数。有关模板的详细信息，请参阅缓存 NetFlow 9 模板。 默认值为 -1 表示缓存大小不受限制。 仅适用于网流 9 数据。
   模板缓存超时（毫秒）	缓存空闲模板的最大毫秒数。未使用超过指定时间的模板将从缓存中逐出。有关模板的详细信息，请参阅缓存 NetFlow 9 模板。 无限期缓存模板的默认值为 -1。 仅适用于网流 9 数据。
   字符集	要处理的消息的字符编码。
   忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。

7. 对于分隔数据，在“数据格式”选项卡上，配置以下属性：

   分隔属性	描述: __________
   分隔符格式类型	分隔符格式类型。使用以下选项之一： 默认 CSV - 包含逗号分隔值的文件。忽略文件中的空行。 RFC4180 CSV - 严格遵循 RFC4180 准则的逗号分隔文件。 MS Excel CSV - Microsoft Excel 逗号分隔的文件。 MySQL CSV - 以逗号分隔的 MySQL 文件。 制表符分隔值 - 包含制表符分隔值的文件。 PostgreSQL CSV - PostgreSQL 逗号分隔的文件。 PostgreSQL Text - PostgreSQL 文本文件。 自定义 - 使用用户定义的分隔符、转义符和引号字符的文件。 多字符分隔 - 使用多个用户定义的字符来分隔字段和行，以及单个用户定义的转义字符和引号字符的文件。
   标题行	指示文件是否包含标题行，以及是否使用标题行。
   允许额外的列	使用标题行处理数据时， 允许处理列数多于标题行中存在的记录。
   额外的列前缀	用于任何其他列的前缀。额外的列使用前缀和顺序递增的整数命名，如下所示：。<prefix><integer> 例如。默认值为 。_extra_1_extra_
   最大记录长度（字符）	记录的最大长度（以字符为单位）。不会读取较长的记录。 此属性可以由数据收集器分析器缓冲区大小限制。有关详细信息，请参阅最大记录大小。
   分隔符字符	自定义分隔符格式的分隔符字符。选择其中一个可用选项或使用“其他”输入自定义字符。 您可以使用 格式 输入 Unicode 控制字符，其中 N 是数字 0-9 或字母 A-F 中的十六进制数字。例如，Enter 可使用空字符作为分隔符，或使用行分隔符作为分隔符。\uNNNN\u0000\u2028 默认值为管道字符 （ | ）。
   多字符字段分隔符	分隔多字符分隔符格式字段的字符。 默认值为两个管道字符（||）。
   多字符线分隔符	为多字符分隔符格式分隔行或记录的字符。 默认值为换行符 （\n）。
   转义字符	自定义或多字符分隔符格式的转义字符。
   引用字符	自定义或多字符分隔符格式的引号字符。
   启用注释	允许对自定义分隔符格式忽略注释的数据。
   注释标记	为自定义分隔符格式启用注释时标记注释的字符。
   忽略空行	允许对自定义分隔符格式忽略空行。
   根字段类型	要使用的根字段类型： 列表映射 - 生成索引数据列表。使您能够使用标准函数来处理数据。用于新管道。 List - 生成一条记录，其中包含一个索引列表，其中包含标头和值的映射。需要使用带分隔符的数据函数来处理数据。仅用于维护在 1.1.0 之前创建的管道。
   要跳过的行	读取数据之前要跳过的行数。
   解析非线性	将指定的字符串常量替换为空值。
   空常量	要替换为空值的字符串常量。
   字符集	要处理的文件的字符编码。
   忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。

8. 对于 JSON 数据，在“数据格式”选项卡上，配置以下属性：

   JSON 属性	描述: __________
   内容	JSON 内容的类型。使用以下选项之一： 对象数组 多个对象
   最大对象长度（字符）	JSON 对象中的最大字符数。 较长的对象将转移到管道以进行错误处理。 此属性可以由数据收集器分析器缓冲区大小限制。有关详细信息，请参阅最大记录大小。
   字符集	要处理的文件的字符编码。
   忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。

9. 对于日志数据，在“数据格式”选项卡上，配置以下属性：

   日志属性	描述: __________
   日志格式	日志文件的格式。使用以下选项之一： 通用日志格式 组合日志格式 阿帕奇错误日志格式 阿帕奇访问日志自定义格式 正则表达式 格罗克图案 日志4j 通用事件格式 日志事件扩展格式
   最大线长	日志行的最大长度。原点截断较长的行。 此属性可以由数据收集器分析器缓冲区大小限制。有关详细信息，请参阅最大记录大小。
   保留原始行	确定如何处理原始日志行。选择此选项可将原始日志行作为字段包含在生成的记录中。 默认情况下，将丢弃原始行。
   字符集	要处理的文件的字符编码。
   忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。

   • 选择“Apache 访问日志自定义格式”时，请使用 Apache 日志格式字符串定义“自定义日志格式”。
   • 选择“正则表达式”时，请输入描述日志格式的正则表达式，然后将要包括的字段映射到每个正则表达式组。
   • 选择“格罗克图案”时，可以使用“格罗克图案定义”字段来定义自定义凹槽图案。您可以在每条线上定义一个图案。

     在“Grok 模式”字段中，输入用于分析日志的模式。您可以使用预定义的凹槽图案，也可以使用在 Grok 图案定义中定义的图案创建自定义凹槽图案。

     有关定义凹槽图案和支持的凹槽图案的详细信息，请参阅定义凹槽图案。

   • 选择 Log4j 时，请定义以下属性：

     Log4j 属性	描述: __________
     解析错误时	确定如何处理无法分析的信息： 跳过并记录错误 - 跳过读取行并记录阶段错误。 跳过，无错误 - 跳过读取该行，不记录错误。 包含为堆栈跟踪 - 包括无法解析为先前读取的日志行的堆栈跟踪的信息。该信息将添加到最后一个有效日志行的消息字段中。
     使用自定义日志格式	允许您定义自定义日志格式。
     自定义日志4J格式	使用 log4j 变量定义自定义日志格式。

10. 对于 protobuf 数据，请在“数据格式”选项卡上，配置以下属性：

    普罗托布夫 房产	描述: __________
    原型描述符文件	要使用的描述符文件 （.desc）。描述符文件必须位于数据收集器资源目录 。$SDC_RESOURCES 有关生成描述符文件的信息，请参阅 Protobuf 数据格式先决条件。有关环境变量的详细信息，请参阅数据收集器环境配置。
    消息类型	读取数据时要使用的消息类型的完全限定名称。 使用以下格式：。<package name>.<message type> 使用描述符文件中定义的消息类型。
    分隔消息	指示消息是否可能包含多个原始缓冲区消息。

11. 对于文本数据，在“数据格式”选项卡上，配置以下属性：

    文本属性	描述: __________
    最大线长	一行允许的最大字符数。较长的行将被截断。 向记录中添加一个布尔字段，以指示它是否被截断。字段名称被截断。 此属性可以由数据收集器分析器缓冲区大小限制。有关详细信息，请参阅最大记录大小。
    使用自定义分隔符	使用自定义分隔符来定义记录而不是换行符。
    自定义分隔符	用于定义记录的一个或多个字符。
    包括自定义分隔符	在记录中包含分隔符字符。
    字符集	要处理的文件的字符编码。
    忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。

12. 对于 XML 数据，在“数据格式”选项卡上，配置以下属性：

    属性	描述: __________
    分隔符元素	用于生成记录的分隔符。省略分隔符可将整个 XML 文档视为一条记录。使用下列方法之一： 根元素正下方的 XML 元素。 使用不带尖括号的 XML 元素名称 （< >）。例如，使用消息而不是<消息>。 一个简化的 XPath 表达式，用于指定要使用的数据。 使用简化的 XPath 表达式访问 XML 文档中更深层次的数据或需要更复杂访问方法的数据。 有关有效语法的详细信息，请参阅简化的 XPath 语法。
    包括字段 X 路径	包括指向字段属性中的每个已分析的 XML 元素和 XML 属性的 XPath。还包括 xmlns 记录标头属性中的每个命名空间。 如果未选中，则记录中不包括此信息。默认情况下，不选择该属性。 注： 仅当您在目标中使用 SDC RPC 数据格式时，字段属性和记录标头属性才会自动写入目标系统。有关使用字段属性和记录头属性以及如何将它们包含在记录中的详细信息，请参阅 字段属性和 记录头属性。
    命名空间	分析 XML 文档时要使用的命名空间前缀和 URI。当正在使用的 XML 元素包含命名空间前缀或 XPath 表达式包含命名空间时，定义命名空间。 有关将命名空间与 XML 元素一起使用的信息，请参阅将 XML 元素与命名空间结合使用。 有关将命名空间与 XPath 表达式配合使用的信息，请参阅将 XPath 表达式与命名空间配合使用。 使用简单或批量编辑模式，单击“添加”图标以添加其他命名空间。
    输出字段属性	在记录中将 XML 属性和命名空间声明作为字段属性包括在内。如果未选中，XML 属性和命名空间声明将作为字段包含在记录中。 注： 仅当您在目标中使用 SDC RPC 数据格式时，字段属性才会自动包含在写入目标系统的记录中。有关使用字段属性的详细信息，请参阅 字段属性。 默认情况下，不选择该属性。
    最大记录长度（字符）	记录中的最大字符数。较长的记录将转移到管道以进行错误处理。 此属性可以由数据收集器分析器缓冲区大小限制。有关详细信息，请参阅最大记录大小。
    字符集	要处理的文件的字符编码。
    忽略控制字符	删除除制表符、换行符和回车符之外的所有 ASCII 控制字符。