经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Apache Kafka » 查看文章
替代Flume——Kafka Connect简介
来源：cnblogs　　作者：独孤风　　时间：2019/8/30 12:35:11　　对本文有异议

file
我们知道过去对于Kafka的定义是分布式，分区化的，带备份机制的日志提交服务。也就是一个分布式的消息队列，这也是他最常见的用法。但是Kafka不止于此，打开最新的官网。

file

我们看到Kafka最新的定义是：Apache Kafka? is?a distributed streaming platform

分布式流处理平台。

file

这里也清晰的描述了Kafka的特点：Kafka用于构建实时数据管道和流式应用程序。它具有水平可扩展性、容错性、速度极快，并在数千家公司投入生产。

所以现在的Kafka已经不仅是一个分布式的消息队列，更是一个流处理平台。这源于它于0.9.0.0和0.10.0.0引入的两个全新的组件Kafka Connect与Kafka Streaming。

Kafka Connect简介

我们知道消息队列必须存在上下游的系统，对消息进行搬入搬出。比如经典的日志分析系统，通过flume读取日志写入kafka，下游由storm进行实时的数据处理。

file

Kafka Connect的作用就是替代Flume，让数据传输这部分工作可以由Kafka Connect来完成。Kafka Connect是一个用于在Apache Kafka和其他系统之间可靠且可靠地传输数据的工具。它可以快速地将大量数据集合移入和移出Kafka。

Kafka Connect的导入作业可以将数据库或从应用程序服务器收集的数据传入到Kafka，导出作业可以将Kafka中的数据传递到查询系统，也可以传输到批处理系统以进行离线分析。

Kafka Connect功能包括：

  • 一个通用的Kafka连接的框架?- Kafka Connect规范化了其他数据系统与Kafka的集成，简化了连接器开发，部署和管理
  • 分布式和独立模式?- 支持大型分布式的管理服务，也支持小型生产环境的部署
  • REST界面?- 通过易用的REST API提交和管理Kafka Connect
  • 自动偏移管理?- 只需从连接器获取一些信息，Kafka Connect就可以自动管理偏移量提交过程，因此连接器开发人员无需担心连接器开发中偏移量提交这部分的开发
  • 默认情况下是分布式和可扩展的?- Kafka Connect构建在现有的组管理协议之上。可以添加扩展集群
  • 流媒体/批处理集成?- 利用Kafka现有的功能，Kafka Connect是桥接流媒体和批处理数据系统的理想解决方案

file

运行Kafka Connect

Kafka Connect目前支持两种运行模式：独立和集群。

独立模式

在独立模式下，只有一个进程，这种更容易设置和使用。但是没有容错功能。

启动：
  1. > bin/connect-standalone.sh config/connect-standalone.properties connector1.properties [connector2.properties ...]
独立模式配置

第一个参数config/connect-standalone.properties是一些基本的配置：

这几个在独立和集群模式下都需要设置：

  1. #bootstrap.servers   kafka集群列表
  2. bootstrap.servers=localhost:9092
  3. #key.converter       key的序列化转换器  比如json的  key.converter=org.apache.kafka.connect.json.JsonConverter
  4. #value.converter     value的序列化转换器
  5. value.converter=org.apache.kafka.connect.json.JsonConverter
  7. #独立模式特有的配置：
  8. #offset.storage.file.filename       用于存储偏移量的文件
  9. offset.storage.file.filename =/home/kafka/connect.offsets
独立模式连接器配置（配置文件）

后面的参数connector1.properties [connector2.properties ...] 可以多个，是连接器配置内容

这里我们配置一个从文件读取数据并存入kafka的配置：

connect-file-sink.properties

  • name?- 连接器的唯一名称。尝试再次使用相同名称注册将失败。

  • connector.class?- 连接器的Java类 此连接器的类的全名或别名。这里我们选择FileStreamSink

  • tasks.max?- 应为此连接器创建的最大任务数。如果连接器无法达到此级别的并行性，则可能会创建更少的任务。

  • key.converter?- （可选）覆盖worker设置的默认密钥转换器。

  • value.converter?- （可选）覆盖worker设置的默认值转换器。

    下面两个必须设置一个：

    • topics?- 以逗号分隔的主题列表，用作此连接器的输入
    • topics.regex?- 用作此连接器输入的主题的Java正则表达式
  1. name=local-file-sink
  2. connector.class=FileStreamSink
  3. tasks.max=1
  4. file=test.sink.txt
  5. topics=connect-test

可以在连接器中配置转换器

需要指定参数：

  • transforms?- 转换的别名列表，指定将应用转换的顺序。
  • transforms.$alias.type?- 转换的完全限定类名。
  • transforms.$alias.$transformationSpecificConfig?转换的配置属性

例如，我们把刚才的文件转换器的内容添加字段

首先设置connect-standalone.properties

  1. key.converter.schemas.enable = false
  2. value.converter.schemas.enable = false

设置connect-file-source.properties

  1. name=local-file-source
  2. connector.class=FileStreamSource
  3. tasks.max=1
  4. file=test.txt
  5. topic=connect-test
  6. transforms=MakeMap, InsertSource
  7. transforms.MakeMap.type=org.apache.kafka.connect.transforms.HoistField$Value
  8. transforms.MakeMap.field=line
  9. transforms.InsertSource.type=org.apache.kafka.connect.transforms.InsertField$Value
  10. transforms.InsertSource.static.field=data_source
  11. transforms.InsertSource.static.value=test-file-source

没有转换前的结果：

  1. "foo"
  2. "bar"
  3. "hello world"

转换后：

  1. {"line":"foo","data_source":"test-file-source"}
  2. {"line":"bar","data_source":"test-file-source"}
  3. {"line":"hello world","data_source":"test-file-source"}

常用转换类型：

  • InsertField - 使用静态数据或记录元数据添加字段
  • ReplaceField - 过滤或重命名字段
  • MaskField - 用类型的有效空值替换字段（0，空字符串等）
  • ValueToKey Value转换为Key
  • HoistField - 将整个事件作为单个字段包装在Struct或Map中
  • ExtractField - 从Struct和Map中提取特定字段，并在结果中仅包含此字段
  • SetSchemaMetadata - 修改架构名称或版本
  • TimestampRouter - 根据原始主题和时间戳修改记录主题
  • RegexRouter - 根据原始主题，替换字符串和正则表达式修改记录主题

集群模式

集群模式下，可以扩展，容错。

启动：
  1. > bin/connect-distributed.sh config/connect-distributed.properties

在集群模式下，Kafka Connect在Kafka主题中存储偏移量，配置和任务状态。

集群模式配置

connect-distributed.properties

  1. #也需要基本的配置
  2. bootstrap.servers=localhost:9092
  3. key.converter=org.apache.kafka.connect.json.JsonConverter
  4. value.converter=org.apache.kafka.connect.json.JsonConverter
  6. #还有一些配置要注意
  7. #group.id（默认connect-cluster） - Connect的组id 请注意，这不得与使用者的组id 冲突
  8. group.id=connect-cluster
  10. #用于存储偏移的主题; 此主题应具有许多分区
  11. offset.storage.topic=connect-offsets
  12. offset.storage.replication.factor=1
  14. #用于存储连接器和任务配置的主题  只能一个分区
  15. config.storage.topic=connect-configs
  16. config.storage.replication.factor=1
  18. #用于存储状态的主题; 此主题可以有多个分区
  19. status.storage.topic=connect-status
  20. status.storage.replication.factor=1

在集群模式下，配置并不会在命令行传进去，而是需要REST API来创建，修改和销毁连接器。

集群模式连接器配置（REST API）

可以配置REST API服务器，支持http与https

  1. listeners=http://localhost:8080,https://localhost:8443

默认情况下，如果未listeners指定，则REST服务器使用HTTP协议在端口8083上运行。

以下是当前支持的REST API：

  • GET /connectors?- 返回活动连接器列表
  • POST /connectors?- 创建一个新的连接器;?请求主体应该是包含字符串name字段的JSON对象和包含config连接器配置参数的对象字段
  • GET /connectors/{name}?- 获取有关特定连接器的信息
  • GET /connectors/{name}/config?- 获取特定连接器的配置参数
  • PUT /connectors/{name}/config?- 更新特定连接器的配置参数
  • GET /connectors/{name}/status?- 获取连接器的当前状态，包括它是否正在运行，失败，暂停等，分配给哪个工作人员，错误信息（如果失败）以及所有任务的状态
  • GET /connectors/{name}/tasks?- 获取当前为连接器运行的任务列表
  • GET /connectors/{name}/tasks/{taskid}/status?- 获取任务的当前状态，包括它是否正在运行，失败，暂停等，分配给哪个工作人员，以及错误信息是否失败
  • PUT /connectors/{name}/pause?- 暂停连接器及其任务，这将停止消息处理，直到恢复连接器
  • PUT /connectors/{name}/resume?- 恢复暂停的连接器（如果连接器未暂停，则不执行任何操作）
  • POST /connectors/{name}/restart?- 重新启动连接器（通常是因为它已经失败）
  • POST /connectors/{name}/tasks/{taskId}/restart?- 重启个别任务（通常因为失败）
  • DELETE /connectors/{name}?- 删除连接器，暂停所有任务并删除其配置

连接器开发指南

kakfa允许开发人员自己去开发一个连接器。

核心概念

要在Kafka和其他系统之间复制数据，用户需要创建一个Connector

Connector有两种形式：

SourceConnectors从另一个系统导入数据，例如，JDBCSourceConnector将关系数据库导入Kafka

SinkConnectors导出数据，例如，HDFSSinkConnector将Kafka主题的内容导出到HDFS文件

和对应的Task：

SourceTask和SinkTask

Task形成输入输出流，开发Task要注意偏移量的问题。

每个流应该是一系列键值记录。还需要定期提交已处理的数据的偏移量，以便在发生故障时，处理可以从上次提交的偏移量恢复。Connector还需要是动态的，实现还负责监视外部系统是否存在任何更改。

开发一个简单的连接器

开发连接器只需要实现两个接口，即Connector和Task。

这里我们简单开发一个FileStreamConnector。

此连接器是为在独立模式下使用，SourceConnector/?SourceTask读取文件的每一行，SinkConnector/?SinkTask每个记录写入一个文件。

连接器示例：

继承SourceConnector，添加字段（要读取的文件名和要将数据发送到的主题）

  1. public class FileStreamSourceConnector extends SourceConnector {
  2.     private String filename;
  3.     private String topic;

定义实际读取数据的类

  1. @Override
  2. public Class<? extends Task> taskClass() {
  3.     return FileStreamSourceTask.class;
  4. }

在FileStreamSourceTask下面定义该类。接下来，我们添加一些标准的生命周期方法，start()和stop()

  1. @Override
  2. public void start(Map<String, String> props) {
  3.     // The complete version includes error handling as well.
  4.     filename = props.get(FILE_CONFIG);
  5.     topic = props.get(TOPIC_CONFIG);
  6. }
  7.  
  8. @Override
  9. public void stop() {
  10.     // Nothing to do since no background monitoring is required.
  11. }

最后，实施的真正核心在于taskConfigs()

  1. @Override
  2. public List<Map<String, String>> taskConfigs(int maxTasks) {
  3.     ArrayList<Map<String, String>> configs = new ArrayList<>();
  4.     // Only one input stream makes sense.
  5.     Map<String, String> config = new HashMap<>();
  6.     if (filename != null)
  7.         config.put(FILE_CONFIG, filename);
  8.     config.put(TOPIC_CONFIG, topic);
  9.     configs.add(config);
  10.     return configs;
  11. }

任务示例：

源任务

实现SourceTask 创建FileStreamSourceTask继承SourceTask

  1. public class FileStreamSourceTask extends SourceTask {
  2.     String filename;
  3.     InputStream stream;
  4.     String topic;
  5.  
  6.     @Override
  7.     public void start(Map<String, String> props) {
  8.         filename = props.get(FileStreamSourceConnector.FILE_CONFIG);
  9.         stream = openOrThrowError(filename);
  10.         topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG);
  11.     }
  12.  
  13.     @Override
  14.     public synchronized void stop() {
  15.         stream.close();
  16.     }

接下来，我们实现任务的主要功能，即poll()从输入系统获取事件并返回以下内容的方法List：

  1. @Override
  2. public List<SourceRecord> poll() throws InterruptedException {
  3.     try {
  4.         ArrayList<SourceRecord> records = new ArrayList<>();
  5.         while (streamValid(stream) && records.isEmpty()) {
  6.             LineAndOffset line = readToNextLine(stream);
  7.             if (line != null) {
  8.                 Map<String, Object> sourcePartition = Collections.singletonMap("filename", filename);
  9.                 Map<String, Object> sourceOffset = Collections.singletonMap("position", streamOffset);
  10.                 records.add(new SourceRecord(sourcePartition, sourceOffset, topic, Schema.STRING_SCHEMA, line));
  11.             } else {
  12.                 Thread.sleep(1);
  13.             }
  14.         }
  15.         return records;
  16.     } catch (IOException e) {
  17.         // Underlying stream was killed, probably as a result of calling stop. Allow to return
  18.         // null, and driving thread will handle any shutdown if necessary.
  19.     }
  20.     return null;
  21. }
接收任务

不像SourceConnector和SinkConnector，SourceTask并SinkTask有非常不同的接口，因为SourceTask采用的是拉接口，并SinkTask使用推接口。两者共享公共生命周期方法，但SinkTask完全不同：

  1. public abstract class SinkTask implements Task {
  2.     public void initialize(SinkTaskContext context) {
  3.         this.context = context;
  4.     }
  5.  
  6.     public abstract void put(Collection<SinkRecord> records);
  7.  
  8.     public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
  9.     }

这是一个简单的例子，它们有简单的结构化数据 - 每一行只是一个字符串。几乎所有实用的连接器都需要具有更复杂数据格式的模式。要创建更复杂的数据，您需要使用Kafka Connect?dataAPI。

  1. Schema schema = SchemaBuilder.struct().name(NAME)
  2.     .field("name", Schema.STRING_SCHEMA)
  3.     .field("age", Schema.INT_SCHEMA)
  4.     .field("admin", new SchemaBuilder.boolean().defaultValue(false).build())
  5.     .build();
  6.  
  7. Struct struct = new Struct(schema)
  8.     .put("name", "Barbara Liskov")
  9.     .put("age", 75);

更多Kafka相关技术文章：

什么是Kafka？
Kafka监控工具汇总
Kafka快速入门
Kafka核心之Consumer
Kafka核心之Producer

更多实时计算,Flink,Kafka等相关技术博文，欢迎关注实时流式计算

file

原文链接：http://www.cnblogs.com/tree1123/p/11434047.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Apache Kafka热门文章

  • Kafka监控工具汇总
  • Kraft模式下Kafka脚本的使用
  • Kafka web console安装
  • kafka_2.11-2.0.0_安装部署
  • ELK日志方案--使用Filebeat收集日志并输出到Kafka
  • Kafka相关内容总结（存储和性能）
  • 一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
  • Kafka使用jmxtrans+influxdb+grafana监控JMX指标
  • Micronaut 微服务中使用 Kafka
  • 什么是Kafka？

    • Apache Kafka推荐文章

  • 异源数据同步 → DataX 为什么要支持 kafka？
  • 解锁强强组合: 使用 Kafka + ClickHouse 快速搭建流数据实时处理平台(DoubleCloud 博客)
  • Kafka 线上性能调优
  • Kafka原理剖析之「位点提交」
  • Kafka 社区KIP-382中文译文（MirrorMaker2/集群复制/高可用/灾难恢复）
  • kafka源码阅读之MacBook Pro M1搭建Kafka2.7版本源码运行环境
  • Kafka干货之「零拷贝」
  • 扫盲Kafka？看这一篇就够了！
  • 从一次Kafka宕机说起（JVM hang）
  • 从Kafka中学习高性能系统如何设计

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号