Kafka
Kafka 插件可让您监控您的 Kafka 事件流处理流程，创建消费者、生产者和主题。 它还允许您连接到 Schema Registry，创建并更新架构。
安装 Kafka 插件
此功能依赖于 Kafka 插件，您需要安装并启用它。
按 Ctrl+Alt+S 打开设置，然后选择 。
打开 Marketplace 选项卡，找到 Kafka 插件，并点击 安装 （如有提示，请重启 IDE）。
使用 Kafka 插件，您可以：
连接到：
如果已安装并启用 Kafka 插件，您可以使用 Kafka 工具窗口（ ）连接到 Kafka 并进行操作。 或者，如果已安装并启用 远程文件系统 或 Zeppelin 插件，您也可以通过 Big Data Tools 工具窗口（ ）访问 Kafka 连接。
连接到 Kafka
使用云服务提供商连接到 Kafka
连接到 Confluent 集群
打开 Kafka 工具窗口： 。
点击
（新建连接）。在 名称 字段中，输入连接名称，以便与其他连接区分开来。
在 配置来源 列表中选择 云 ，然后在 提供商 列表中选择 Confluent。
转到 https://confluent.cloud/home。 在页面右侧，点击设置菜单，选择 环境 ，并选择您的集群，然后选择 。
在 为您的客户端复制配置代码片段 部分中，提供 Kafka API 密钥，并点击 复制。
返回您的 IDE，并将复制的属性粘贴到 配置 字段中。
完成设置后，点击 测试连接 以确保所有配置参数正确。 然后点击 确定。
可选，您可以设置：
启用连接 ：如果要禁用此连接，请清除此复选框。 默认情况下，新建的连接处于启用状态。
按项目 ：选择此项可仅为当前项目启用这些连接设置。 如果希望在其他项目中也可见此连接，请清除此复选框。
连接到 AWS MSK 集群
打开 Kafka 工具窗口： 。
点击
（新建连接）。在 名称 字段中，输入连接名称，以便与其他连接区分开来。
在 配置来源 列表中选择 云 ，然后在 提供商 列表中选择 AWS MSK。
在 引导服务器 字段中，输入 Kafka Broker 的 URL，或以逗号分隔的 URL 列表。
在 AWS 身份验证 列表中，选择身份验证方法。
默认凭证提供程序链 ：使用默认提供程序链中的凭据。 有关该链的详细信息，请参阅 Using the Default Credential Provider Chain。
来自凭证文件的配置文件 ：从您的 文件中选择一个配置文件。
显式访问密钥和秘密访问密钥 ：手动输入您的凭据。
可选，您可以 连接到 Schema Registry。
如果希望在连接到 Kafka 时使用 SSH 隧道，请选择 启用隧道 ，并在 SSH 配置 列表中选择一个 SSH 配置，或创建一个新配置。
完成设置后，点击 测试连接 以确保所有配置参数正确。 然后点击 确定。
可选，您可以设置：
启用连接 ：如果要禁用此连接，请清除此复选框。 默认情况下，新建的连接处于启用状态。
按项目 ：选择此项可仅为当前项目启用这些连接设置。 如果希望在其他项目中也可见此连接，请清除此复选框。
连接到自定义 Kafka 服务器
打开 Kafka 工具窗口： 。
点击
（新建连接）。在 名称 字段中，输入连接名称，以便与其他连接区分开来。
在 配置来源 列表中选择 自定义。
在 引导服务器 字段中，输入 Kafka Broker 的 URL，或以逗号分隔的 URL 列表。
在 身份验证 下，选择一种身份验证方法：
无 ：在不进行身份验证的情况下连接。
SASL ：选择一种 SASL 机制（Plain、SCRAM-SHA-256、SCRAM-SHA-512 或 Kerberos ），并提供您的用户名和密码。
SSL
如果希望验证 Broker 主机名与 Broker 证书中的主机名是否匹配，请选择 验证服务器主机名。 清除此复选框等同于添加
ssl.endpoint.identification.algorithm=属性。在 信任库位置 中，提供 SSL 信任库位置的路径（
ssl.truststore.location属性）。在 信任库密码 中，提供 SSL 信任库密码的路径（
ssl.truststore.password属性）。选择 使用密钥库进行客户端身份验证 ，并为 密钥库位置 （
ssl.keystore.location）、 密钥库密码 （ssl.keystore.password）和 密钥密码 （ssl.key.password）提供值。
AWS IAM ：对 Amazon MSK 使用 AWS IAM。 在 AWS 身份验证 列表中，选择以下之一：
默认凭证提供程序链 ：使用默认提供程序链中的凭据。 有关该链的详细信息，请参阅 Using the Default Credential Provider Chain。
来自凭证文件的配置文件 ：从您的 文件中选择一个配置文件。
显式访问密钥和秘密访问密钥 ：手动输入您的凭据。
可选，您可以 连接到 Schema Registry。
如果希望在连接到 Kafka 时使用 SSH 隧道，请选择 启用隧道 ，并在 SSH 配置 列表中选择一个 SSH 配置，或创建一个新配置。
完成设置后，点击 测试连接 以确保所有配置参数正确。 然后点击 确定。
可选，您可以设置：
启用连接 ：如果要禁用此连接，请清除此复选框。 默认情况下，新建的连接处于启用状态。
按项目 ：选择此项可仅为当前项目启用这些连接设置。 如果希望在其他项目中也可见此连接，请清除此复选框。
使用属性连接到 Kafka
打开 Kafka 工具窗口： 。
点击
（新建连接）。在 名称 字段中，输入连接名称，以便与其他连接区分开来。
在 配置来源 列表中选择 属性。
在 引导服务器 字段中，输入 Kafka Broker 的 URL，或以逗号分隔的 URL 列表。
选择提供 Kafka Broker 配置属性的方式：
隐式 ：粘贴提供的配置属性。 或者，您可以使用 PyCharm 提供的代码补全和快速文档手动输入它们。
来自文件 ：选择属性文件。
可选，您可以 连接到 Schema Registry。
如果希望在连接到 Kafka 时使用 SSH 隧道，请选择 启用隧道 ，并在 SSH 配置 列表中选择一个 SSH 配置，或创建一个新配置。
完成设置后，点击 测试连接 以确保所有配置参数正确。 然后点击 确定。
可选，您可以设置：
启用连接 ：如果要禁用此连接，请清除此复选框。 默认情况下，新建的连接处于启用状态。
按项目 ：选择此项可仅为当前项目启用这些连接设置。 如果希望在其他项目中也可见此连接，请清除此复选框。
与 Kafka 服务器建立连接后， Kafka 工具窗口中会出现一个带有此连接的新选项卡。 您可以使用它来 生产和 消费数据， 创建并删除主题。 如果您 已连接到 Schema Registry ，您还可以查看、创建和更新架构。
在 Kafka 工具窗口的任一选项卡中点击 
，可重命名、删除、禁用或刷新该连接，或修改其设置。
所有集群主题都会显示在 主题 部分。 您可以点击 
以仅显示收藏的主题，或点击 
以显示或隐藏内部主题。 点击任意主题可获取其更多详细信息，例如分区、配置和架构信息。
创建主题
打开 Kafka 工具窗口： 。
点击
（新建连接）。选择 主题 ，并点击
（或按 Alt+Insert）。为新主题命名，指定分区数和复制因子，然后点击 确定。
从主题中删除记录
打开 Kafka 工具窗口： 。
点击
（新建连接）。在 主题 下，右键点击某个主题并选择 清除主题 （或点击其左侧的
）。 点击 确定 以确认删除。
生产并消费数据
生产数据
打开 Kafka 工具窗口： 。
点击
（新建连接）。选择一个 Kafka 连接并点击 生产者。
这将在新的编辑器选项卡中打开一个生产者。
在 主题 列表中，选择要写入消息的主题。
在 键 和 值 下，选择消息键和值。
如果您 已连接到 Schema Registry ，您可以选择 模式注册表 以根据所选架构检查已发送的数据。
您可以生成随机值：
点击
，可基于所选类型生成随机值。 这包括基于所选 Schema Registry 生成完整的 JSON 对象。若要更灵活地生成随机值，请使用
${random...}变量。 在编辑 JSON、Avro 或 Protobuf 文件时，在值字段中开始输入random可查看可能的随机值的自动补全选项。 例如，您可以使用"${random.integer(1,10)}"生成 1 到 10 之间的随机整数。
在 头 下，提供任意自定义标头。 如果它们是 JSON 或 CSV 格式，您可以将其粘贴到此部分。
在 流程 下，您可以控制记录流：
在 每次记录数 中，若要同时发送多条记录，请输入一个数字。
如果希望随机生成记录数据，请选择 生成随机键 和 生成随机值。
以毫秒为单位设置 间隔。
如果希望在达到指定记录数或经过指定时间后让生产者停止发送消息，请提供 停止条件。
在 选项 下，提供其他选项：
分区 ：指定必须发送记录的主题分区。 如果未指定，将使用默认逻辑：生产者取键的哈希对分区数取模。
压缩 ：选择生产者生成的数据的压缩类型： 无、 Gzip、 Snappy、 Lz4 或 Zstd。
幂等性 ：如果希望确保每条消息在流中只写入一份，请选择此项。
ACK ：如果希望领导者将记录写入其本地日志并在不等待所有追随者完全确认的情况下进行响应，请选择 Leader。 选择 全部 ，使领导者等待完整的同步副本集合确认该记录。 保留 无 ，使生产者不等待来自服务器的任何确认。
点击 生产。
随后，您可以在 数据 选项卡中点击任意记录以显示其详细信息。 您也可以点击 
以启用统计信息。
消费数据
打开 Kafka 工具窗口： 。
点击
（新建连接）。选择一个 Kafka 连接并点击 消费者。
这将在新的编辑器选项卡中打开一个消费者。
在 主题 列表中，选择要订阅的主题。
在 键 和 值 下，选择您将要消费的记录的键和值的数据类型。
使用 范围与筛选器 缩小要消费的数据范围：
在 起点 列表中，选择要开始消费数据的时段或偏移量。 选择 从开头 以获取该主题的所有记录。
在 限制 列表中，选择何时停止接收数据，例如当主题达到某个记录数时。
使用 筛选器 按键、值或标头中的子字符串筛选记录。
在 其他 下，配置其他参数：
在 分区 框中，输入分区 ID 或以逗号分隔的 ID 列表，仅从特定分区获取记录。
在 消费者组 列表中，如果希望将新消费者加入其中，请选择一个消费者组。
点击 开始消费。
随后，您可以在 数据 选项卡中点击任意记录以显示其详细信息。 您也可以点击 以启用统计信息。
导出数据
您可以以 CSV、TSV 或 JSON 格式下载已生产或已消费的数据。
并选择 保存生产者或消费者预设
如果您经常使用相同的键、值、标头或其他参数生产或消费数据，您可以将它们保存为预设。 随后，您可以重用预设以快速创建生产者或消费者。
在 Kafka 工具窗口中，点击 生产者 或 消费者。
指定所需参数，然后在生产者或消费者创建表单的顶部点击
（保存预设）。
这些参数会保存为一个预设，可在 预设 选项卡中使用。 点击某个预设以应用它。
使用 Schema Registry
生产者和消费者可以使用架构来验证并确保其记录的键和值的一致性。 Kafka 插件与 Schema Registry 集成，支持 Avro、Protobuf 和 JSON 架构。 它使您能够：
连接到 Schema Registry
创建、更新、删除并克隆架构
以原始格式或树形视图预览架构
比较架构版本
删除架构版本
连接到 Schema Registry
如果您使用 Confluent ，您可以将 Broker 和 Schema Registry 的属性都粘贴到 配置 字段中。
否则，展开 模式注册表 部分并选择提供商： Confluent 或 Glue。
URL ：输入 Schema Registry URL。
配置来源 ：选择提供连接参数的方式：
自定义 ：选择身份验证方法并提供凭据。
如果您希望使用不同于 Kafka Broker 的 SSL 设置，请清除 使用代理 SSL 设置 复选框并提供信任库的路径。
属性 ：粘贴提供的配置属性。 或者，您可以使用 PyCharm 提供的代码补全和快速文档手动输入属性。
区域 ：选择 Schema Registry 区域。
AWS 身份验证 ：选择身份验证方法：
默认凭证提供程序链 ：使用默认提供程序链中的凭据。 有关该链的详细信息，请参阅 Using the Default Credential Provider Chain。
来自凭证文件的配置文件 ：从您的 文件中选择一个配置文件。
显式访问密钥和秘密访问密钥 ：手动输入您的凭据。
注册表名称 ：输入您希望连接的 Schema Registry 的名称，或点击
从列表中选择。
完成设置后，点击 测试连接 以确保所有配置参数正确。 然后点击 确定。
创建架构
打开 Kafka 工具窗口： 。
点击
（新建连接）。选择 模式注册表 并点击
（或按 Alt+Insert）。在 格式 列表中，选择架构格式：Avro、Protobuf 或 JSON。
在 策略 列表中，选择 naming strategy ，并根据所选策略设置名称后缀或选择一个主题。 或者，选择 自定义名称 并输入任意名称。
您可以在树形视图和原始视图中预览架构。
比较架构版本
连接到 Schema Registry 后，在 模式注册表 下选择一个架构。
切换到 原始视图 并点击 比较。 仅当架构具有多个版本时，该按钮才可用。
删除架构版本
如果某个架构有多个版本，您可以删除特定版本。 Schema Registry 支持 两种类型的删除 ：软（当删除版本后，架构元数据和 ID 不会从注册表中移除）和硬（会移除所有元数据，包括架构 ID）。 能否选择取决于您使用的是 Confluent 还是 AWS Glue Schema Registry：
在 Confluent Schema Registry 中，默认使用软删除。 您可以选择使用硬删除，方法是选中 永久删除 复选框。
AWS Glue Schema Registry 始终使用硬删除。
在 模式注册表 下，选择一个架构。
在其右侧，点击
并选择 删除版本。