kafka学习--使用kafka conect操作kafka connector

1. 配置kafka connectors

kafka connectors配置是简单的键值映射。对于独立模式，这些在属性文件中定义，并传递到命令行上的kafka Connect进程。在分布式模式下，它们将包含在JSON有效负载中，用于创建(或修改)kafka connectors的请求。

大多数配置都依赖于kafka connectors，因此不能在这里列出它们。然而，有几个常见的选择:

name ： kafka connectors的唯一名称。试图用相同的名称重新注册将失败。

connector.class：连接器的Java类

tasks.max: 应该为这个连接器创建的最大任务数。如果连接器不能达到这种并行度，它可能会创建更少的任务。

key.converter: 转换器(可选)覆盖工作人员设置的默认密钥转换器。

value.converter: 转换器-(可选)覆盖工作程序设置的默认值转换器。

class配置支持几种格式:此连接器的类的全名或别名。如果连接器是org.apache.kafka.connect.file.FileStreamSinkConnector。您可以指定这个全名，也可以使用FileStreamSink或FileStreamSinkConnector来缩短配置。

Sink connectors 也有一些额外的选项来控制它们的输入。每个接收器连接器必须设置以下之一:

topics: 使用逗号分隔的主题列表作为此连接器的输入

topics.regex: 用作此连接器输入的主题的Java正则表达式

对于任何其他选项，您应该参考连接器的文档。

2.配置kafka connectors transforms

kafka connectors可以配置transformations来进行轻量级的一次性消息修改。它们可以方便地进行数据按摩和事件路由。

A transformation chain can be specified in the connector configuration.

transforms :   List of aliases for the transformation, specifying the order in which the transformations will be applied.
transforms.$alias.type ：  Fully qualified class name for the transformation.
transforms.$alias.$transformationSpecificConfig：  Configuration properties for the transformation
Kafka Connect包含了几个广泛适用的数据和路由transformations：

InsertField：使用静态数据或记录元数据添加字段
ReplaceField：筛选或重命名字段
MaskField：用类型(0、空字符串等)的有效空值替换字段
ValueToKey
HoistField:   将整个事件包装为struct或map中的单个字段
ExtractField:  从Struct和Map中提取特定字段，并在结果中只包含该字段
SetSchemaMetadata:  修改Schema名称或版本
TimestampRouter： Modify the topic of a record based on original topic and timestamp. Useful when using a sink that needs to write to different tables or indexes based on timestamps
RegexRouter - modify the topic of a record based on original topic, replacement string and a regular expression

3.使用kafka connect rest api 启动kafka connectors

由于Kafka Connect打算作为服务运行，所以它还提供了一个用于管理连接器的REST API。默认情况下，此服务在端口8083上运行。以下是目前支持的端点:

GET /connectors - return a list of active connectors
POST /connectors - create a new connector; the request body should be a JSON object containing a string name field and an object config field with the connector configuration parameters
GET /connectors/{name} - get information about a specific connector
GET /connectors/{name}/config - get the configuration parameters for a specific connector
PUT /connectors/{name}/config - update the configuration parameters for a specific connector
GET /connectors/{name}/status - get current status of the connector, including if it is running, failed, paused, etc., which worker it is assigned to, error information if it has failed, and the state of all its tasks
GET /connectors/{name}/tasks - get a list of tasks currently running for a connector
GET /connectors/{name}/tasks/{taskid}/status - get current status of the task, including if it is running, failed, paused, etc., which worker it is assigned to, and error information if it has failed
PUT /connectors/{name}/pause - pause the connector and its tasks, which stops message processing until the connector is resumed
PUT /connectors/{name}/resume - resume a paused connector (or do nothing if the connector is not paused)
POST /connectors/{name}/restart - restart a connector (typically because it has failed)
POST /connectors/{name}/tasks/{taskId}/restart - restart an individual task (typically because it has failed)
DELETE /connectors/{name} - delete a connector, halting all tasks and deleting its configuration

Kafka Connect also provides a REST API for getting information about connector plugins:

GET /connector-plugins- return a list of connector plugins installed in the Kafka Connect cluster. Note that the API only checks for connectors on the worker that handles the request, which means you may see inconsistent results, especially during a rolling upgrade if you add new connector jars
PUT /connector-plugins/{connector-type}/config/validate - validate the provided configuration values against the configuration definition. This API performs per config validation, returns suggested values and error messages during validation.

kafka学习--使用kafka conect操作kafka connector相关推荐

Kafka学习笔记（3）----Kafka的数据复制（Replica）与Failover
1. CAP理论 1.1 Cosistency(一致性) 通过某个节点的写操作结果对后面通过其他节点的读操作可见. 如果更新数据后,并发访问的情况下可立即感知该更新,称为强一致性如果允许之后部分或全 ...
Kafka学习笔记（八）Kafka消费者
版权声明:本文为转载文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明. 原文链接:https://blog.csdn.net/weixin_39468305/articl ...
python使用kafka原理详解_Python操作Kafka原理及使用详解
Python操作Kafka原理及使用详解一.什么是Kafka Kafka是一个分布式流处理系统,流处理系统使它可以像消息队列一样publish或者subscribe消息,分布式提供了容错性,并发处理 ...
Kafka学习笔记-Java简单操作
Maven依赖包: [plain] view plaincopy <dependency> <groupId>org.apache.kafka</groupId> ...
Kafka学习总结（1）——Kafka入门简介
Kafka是分布式发布-订阅消息系统.它最初由LinkedIn公司开发,之后成为Apache项目的一部分.Kafka是一个分布式的,可划分的,冗余备份的持久性的日志服务.它主要用于处理活跃的流式数据. ...
kafka学习（六）：kafka应用场景
消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构.目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ ...
kafka学习（一）初识kafka
本文借鉴:再过半小时,你就能明白kafka的工作原理了(特此感谢!) 一.简介定义:kafka是一个分布式,基于zookeeper协调的发布/订阅模式的消息系统,本质是一个MQ(消息队列Messag ...
Kafka学习记录(三)——Broker
Kafka学习记录(三)--Broker 目录 Kafka学习记录(三)--Broker 对应课程 Zookeeper存储的Kafka信息 Broker总体工作流程 Broker的服役和退役 Kafk ...
Kafka学习记录(四)——消费者
Kafka学习记录(四)--消费者目录 Kafka学习记录(四)--消费者对应课程 Kafka消费者工作流程消费方式和流程消费者组原理消费者组初始化流程消费者组详细消费流程重要参数 ka ...

kafka学习--使用kafka conect操作kafka connector

kafka学习--使用kafka conect操作kafka connector相关推荐

最新文章

热门文章