如何为fast-data-dev开发自定义连接器：完整开发与集成教程

如何为fast-data-dev开发自定义连接器：完整开发与集成教程

【免费下载链接】fast-data-devKafka Docker for development. Kafka, Zookeeper, Schema Registry, Kafka-Connect, , 20+ connectors项目地址: https://gitcode.com/gh_mirrors/fa/fast-data-dev

fast-data-dev是一个功能强大的Kafka Docker开发环境，集成了Kafka、Zookeeper、Schema Registry和20多种预配置连接器，让开发者能够快速搭建和测试流数据处理管道。本文将详细介绍如何为fast-data-dev开发自定义连接器，从环境准备到集成部署，帮助你轻松扩展数据接入能力。

1. 了解fast-data-dev连接器架构

在开始开发前，先了解fast-data-dev的连接器管理机制。fast-data-dev通过Kafka Connect框架管理连接器，所有连接器插件存放在指定目录中，并通过配置文件和脚本进行加载和启用。

连接器相关的核心配置和脚本位于以下路径：

• 连接器配置模板：filesystem/usr/local/share/lensesio/etc/supervisord.templates.d/
• 连接器管理脚本：setup-and-run.sh
• 示例连接器脚本：filesystem/usr/local/bin/logs-to-kafka.sh

2. 开发环境准备

2.1 克隆项目仓库

首先，克隆fast-data-dev项目到本地：

git clone https://gitcode.com/gh_mirrors/fa/fast-data-dev cd fast-data-dev

2.2 了解连接器目录结构

fast-data-dev的连接器存放于以下目录：

• 内置连接器：/opt/lensesio/connectors/stream-reactor/
• 第三方连接器：/opt/lensesio/connectors/third-party/
• 运行时连接器：/var/run/connect/connectors/

在setup-and-run.sh脚本中，通过环境变量CONNECT_PLUGIN_PATH指定连接器加载路径：

export CONNECT_PLUGIN_PATH=${CONNECT_PLUGIN_PATH:-/var/run/connect/connectors/stream-reactor,/var/run/connect/connectors/third-party,/connectors}

3. 自定义连接器开发步骤

3.1 创建连接器项目

使用Maven或Gradle创建一个 Kafka Connect 连接器项目，项目结构如下：

my-connector/ ├── src/ │ ├── main/ │ │ ├── java/ │ │ │ └── com/ │ │ │ └── example/ │ │ │ └── connect/ │ │ │ ├── MySourceConnector.java │ │ │ ├── MySourceTask.java │ │ │ └── MyConnectorConfig.java │ │ └── resources/ │ │ └── META-INF/ │ │ └── services/ │ │ └── org.apache.kafka.connect.connector.Connector │ └── test/ │ └── java/ │ └── com/ │ └── example/ │ └── connect/ │ └── MyConnectorTest.java ├── pom.xml (或 build.gradle) └── README.md

3.2 实现连接器核心类

3.2.1 配置类（MyConnectorConfig）

定义连接器配置参数，如服务器地址、用户名、密码等：

public class MyConnectorConfig extends AbstractConfig { public static final String SERVER_URL_CONFIG = "server.url"; private static final String SERVER_URL_DOC = "The URL of the server to connect to"; public MyConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) { super(config, parsedConfig); } public MyConnectorConfig(Map<String, String> parsedConfig) { this(conf(), parsedConfig); } public static ConfigDef conf() { return new ConfigDef() .define(SERVER_URL_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, SERVER_URL_DOC); } public String getServerUrl() { return getString(SERVER_URL_CONFIG); } }

3.2.2 连接器类（MySourceConnector）

实现连接器主类，负责初始化和任务分配：

public class MySourceConnector extends SourceConnector { private MyConnectorConfig config; @Override public void start(Map<String, String> props) { config = new MyConnectorConfig(props); } @Override public Class<? extends Task> taskClass() { return MySourceTask.class; } @Override public List<Map<String, String>> taskConfigs(int maxTasks) { List<Map<String, String>> taskConfigs = new ArrayList<>(); Map<String, String> taskProps = new HashMap<>(); taskProps.put(MyConnectorConfig.SERVER_URL_CONFIG, config.getServerUrl()); for (int i = 0; i < maxTasks; i++) { taskConfigs.add(taskProps); } return taskConfigs; } @Override public void stop() {} @Override public ConfigDef config() { return MyConnectorConfig.conf(); } @Override public String version() { return "1.0.0"; } }

3.2.3 任务类（MySourceTask）

实现数据读取和发送逻辑：

public class MySourceTask extends SourceTask { private MyConnectorConfig config; private HttpClient httpClient; @Override public void start(Map<String, String> props) { config = new MyConnectorConfig(props); httpClient = HttpClient.newHttpClient(); } @Override public List<SourceRecord> poll() throws InterruptedException { List<SourceRecord> records = new ArrayList<>(); // 从数据源读取数据并转换为SourceRecord // 示例：从HTTP API获取数据 HttpRequest request = HttpRequest.newBuilder() .uri(URI.create(config.getServerUrl())) .build(); httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()) .thenApply(HttpResponse::body) .thenAccept(body -> { Map<String, String> sourcePartition = Collections.singletonMap("server", config.getServerUrl()); Map<String, Long> sourceOffset = Collections.singletonMap("offset", System.currentTimeMillis()); records.add(new SourceRecord( sourcePartition, sourceOffset, "my_topic", // 目标Kafka主题 Schema.STRING_SCHEMA, // 键 schema "key", // 键 Schema.STRING_SCHEMA, // 值 schema body // 值 )); }).join(); Thread.sleep(1000); // 每秒轮询一次 return records; } @Override public void stop() { httpClient.close(); } @Override public String version() { return "1.0.0"; } }

3.3 打包连接器

使用Maven打包连接器为JAR文件：

mvn clean package

生成的JAR文件位于target/目录下，如my-connector-1.0.0.jar。

4. 集成自定义连接器到fast-data-dev

4.1 准备连接器目录

在fast-data-dev项目中创建自定义连接器目录：

mkdir -p connectors/my-connector cp target/my-connector-1.0.0.jar connectors/my-connector/

4.2 配置连接器加载

修改setup-and-run.sh脚本，将自定义连接器目录添加到CONNECT_PLUGIN_PATH：

export CONNECT_PLUGIN_PATH=${CONNECT_PLUGIN_PATH:-/var/run/connect/connectors/stream-reactor,/var/run/connect/connectors/third-party,/connectors,/connectors/my-connector}

4.3 创建连接器配置脚本

参考filesystem/usr/local/bin/logs-to-kafka.sh，创建连接器配置脚本filesystem/usr/local/bin/my-connector.sh：

#!/bin/bash CONNECT_PORT=${CONNECT_PORT:-8083} cat <<EOF >/tmp/connector-my { "name": "my-connector", "config": { "connector.class": "com.example.connect.MySourceConnector", "tasks.max": "1", "topic": "my_topic", "server.url": "http://example.com/api/data" } } EOF curl -vs --stderr - -X POST -H "Content-Type: application/json" \ --data @/tmp/connector-my "http://127.0.0.1:$CONNECT_PORT/connectors" rm /tmp/connector-my

4.4 修改Supervisor配置

创建Supervisor配置文件filesystem/usr/local/share/lensesio/etc/supervisord.templates.d/09-my-connector.conf，确保连接器在fast-data-dev启动时自动运行：

[program:my-connector] command=bash -c '/usr/local/bin/my-connector.sh' autostart=true autorestart=true startretries=3 stderr_logfile=/var/log/my-connector.err.log stdout_logfile=/var/log/my-connector.out.log user=root

5. 构建和测试自定义连接器

5.1 构建Docker镜像

docker build -t fast-data-dev-custom .

5.2 启动容器并测试

docker run -p 2181:2181 -p 9092:9092 -p 8081:8081 -p 8083:8083 -p 8082:8082 -p 80:80 fast-data-dev-custom

访问Kafka Connect UI（http://localhost:8083），检查自定义连接器是否成功加载和运行。

5.3 验证数据

使用Kafka命令行工具消费主题数据，验证连接器是否正常工作：

docker exec -it <container_id> kafka-console-consumer --bootstrap-server localhost:9092 --topic my_topic --from-beginning

6. 常见问题与解决方案

6.1 连接器加载失败

检查连接器JAR文件是否正确放置在CONNECT_PLUGIN_PATH指定的目录中，确保连接器类名在META-INF/services/org.apache.kafka.connect.connector.Connector文件中正确配置。

6.2 权限问题

确保连接器脚本和配置文件具有可执行权限，可使用以下命令添加权限：

chmod +x filesystem/usr/local/bin/my-connector.sh

6.3 依赖冲突

如果连接器依赖与fast-data-dev内置依赖冲突，可使用maven-shade-plugin重新打包连接器，修改依赖包的包名。

7. 总结

通过本文的步骤，你可以轻松开发并集成自定义连接器到fast-data-dev环境中，扩展其数据接入能力。fast-data-dev提供了灵活的连接器管理机制，通过简单的配置和脚本修改，即可实现自定义连接器的无缝集成。

希望本文对你开发fast-data-dev自定义连接器有所帮助！如有任何问题，欢迎查阅项目文档或提交issue。

【免费下载链接】fast-data-devKafka Docker for development. Kafka, Zookeeper, Schema Registry, Kafka-Connect, , 20+ connectors项目地址: https://gitcode.com/gh_mirrors/fa/fast-data-dev