Kafka Connect用户指南

半兽人 发表于: 2016-08-22   最后更新时间: 2016-10-27  
  •   168 订阅,8045 游览

8.2 用户指南

提供了一个快速入门的例子,运行一个单机版的Kafka Connect。本节更详细的介绍如何配置,运行和管理Kafka Connect。

运行Kafka Connect

Kafka Connect目前支持两种执行模式:独立(单进程)和分布式。在独立模式下,所有的工作都在一个单进程中进行的。这样易于配置,在一些情况下,只有一个在工作是好的(例如,收集日志文件),但它不会从kafka Connection的功能受益,如容错。通过下面的命令开始一个单进程的例子:

> bin/connect-standalone.sh config/connect-standalone.properties connector1.properties [connector2.properties ...]

第一个参数是worker(工人)的配置,这包括 Kafka连接的参数设置,序列化格式,以及频繁地提交offset(偏移量)。本节提供的例子用的是默认的配置 conf/server.properties。其余的参数是connector(连接器)配置文件。你可以配置你需要的,但是所有的执行都在同一个进程(在不同的线程)。分布式的模式会自动平衡。允许你动态的扩展(或缩减),并在执行任务期间和配置、偏移量提交中提供容错保障,非常类似于独立模式:

bin/connect-distributed.sh config/connect-distributed.properties

在不同的类中,配置参数定义了Kafka Connect如何处理,哪里存储配置,如何分配work,哪里存储offset和任务状态。在分布式模式中,Kafka Connect在topic中存储offset,配置和任务状态。建议手动创建offset的topic,可以自己来定义需要的分区数和副本数。如果启动Kafka Connect时还没有创建topic,那么topic将自动创建(使用默认的分区和副本),这可能不是最合适的(因为kafka可不知道业务需要,只能根据默认参数创建)。特别是以下配置参数尤为关键,启动集群之前设置:

  • group.id (默认connect-cluster) - Connect cluster group使用唯一的名称;注意这不能和consumer group ID(消费者组)冲突。

  • config.storage.topic (默认connect-configs) - topic用于存储connector和任务配置;注意,这应该是一个单个的partition,多副本的topic。你需要手动创建这个topic,以确保是单个partition(自动创建的可能会有多个partition)。

  • offset.storage.topic (默认 connect-offsets) - topic用于存储offsets;这个topic应该配置多个partition和副本。

  • status.storage.topic (默认 connect-status) - topic 用于存储状态;这个topic 可以有多个partitions和副本

注意,在分布式模式中,connector(连接器)配置不能使用命令行。要使用下面介绍的REST API来创建,修改和销毁connector。

配置连接器(connector)

Connector的配置是简单的key-value映射。对于独立模式,这些都是在属性文件中定义,并通过在命令行上的Connect处理。在分布式模式,JSON负载connector的创建(或修改)请求。大多数配置都是依赖的connector,有几个常见的选项:

  • name - 连接器唯一的名称,不能重复。
  • connector.calss - 连接器的Java类。
  • tasks.max - 连接器创建任务的最大数。
  • connector.class配置支持多种格式:全名或连接器类的别名。比如连接器是org.apache.kafka.connect.file.FileStreamSinkConnector,你可以指定全名,也可以使用FileStreamSinkFileStreamSinkConnector。Sink connector也有一个额外的选项来控制它们的输入:
  • topics - 作为连接器的输入的topic列表。

对于其他的选项,你可以查看连接器的文档。

REST API

由于Kafka Connect的目的是作为一个服务运行,提供了一个用于管理connector的REST API。默认情况下,此服务的端口是8083。以下是当前支持的终端入口:

  • GET /connectors - 返回活跃的connector列表
  • POST /connectors - 创建一个新的connector;请求的主体是一个包含字符串name字段和对象config字段(connector的配置参数)的JSON对象。
  • GET /connectors/{name} - 获取指定connector的信息
  • GET /connectors/{name}/config - 获取指定connector的配置参数
  • PUT /connectors/{name}/config - 更新指定connector的配置参数
  • GET /connectors/{name}/status - 获取connector的当前状态,包括它是否正在运行,失败,暂停等。
  • GET /connectors/{name}/tasks - 获取当前正在运行的connector的任务列表。
  • GET /connectors/{name}/tasks/{taskid}/status - 获取任务的当前状态,包括是否是运行中的,失败的,暂停的等,
  • PUT /connectors/{name}/pause - 暂停连接器和它的任务,停止消息处理,直到connector恢复。
  • PUT /connectors/{name}/resume - 恢复暂停的connector(如果connector没有暂停,则什么都不做)
  • POST /connectors/{name}/restart - 重启connector(connector已故障)
  • POST /connectors/{name}/tasks/{taskId}/restart - 重启单个任务 (通常这个任务已失败)
  • DELETE /connectors/{name} - 删除connector, 停止所有的任务并删除其配置

Kafka Connector还提供了获取有关connector plugins信息的REST API:

  • GET /connector-plugins- 返回已在Kafka Connect集群安装的connector plugin列表。请注意,API仅验证处理请求的worker的connector。这以为着你可能看不不一致的结果,特别是在滚动升级的时候(添加新的connector jar)

  • PUT /connector-plugins/{connector-type}/config/validate - 对提供的配置值进行验证,执行对每个配置验证,返回验证的建议值和错误信息。







发表于: 1年前   最后更新时间: 1年前   游览量:8045
上一条: Kafka Connect的用法
下一条: Kafka Connector开发指南
评论…

  • 评论…
    • in this conversation
      提问