Eijux：/* 开发指南【？？？】 */

2021-05-20T04:27:01Z

开发指南【？？？】

←上一版本		2021年5月20日 (四) 12:27的版本
第144行：		第144行：

	# 通过创建一个继承 '''SourceConnector''' 的类开始，增加一个字段存储解析的配置信息（文件名读取和发送数据到topic）：		# 通过创建一个继承 '''SourceConnector''' 的类开始，增加一个字段存储解析的配置信息（文件名读取和发送数据到topic）：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	public class FileStreamSourceConnector extends SourceConnector {		public class FileStreamSourceConnector extends SourceConnector {
	private String filename;		private String filename;
第150行：		第150行：
	</syntaxhighlight>		</syntaxhighlight>
	# 最简单的方法是 '''getTaskClass()'''，它定义了在工作进程中实例化的实际读取数据的类：		# 最简单的方法是 '''getTaskClass()'''，它定义了在工作进程中实例化的实际读取数据的类：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	@Override		@Override
	public Class<? extends Task> getTaskClass() {		public Class<? extends Task> getTaskClass() {
第157行：		第157行：
	</syntaxhighlight>		</syntaxhighlight>
	# 定义 '''FileStreamSourceTask''' 类，接下来，我们增加一些标准的生命周期的方法，start()和stop()：		# 定义 '''FileStreamSourceTask''' 类，接下来，我们增加一些标准的生命周期的方法，start()和stop()：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	@Override		@Override
	public void start(Map<String, String> props) {		public void start(Map<String, String> props) {
第171行：		第171行：
	</syntaxhighlight>		</syntaxhighlight>
	# 最后，是实现真正核心的 '''getTaskConfigs()'''。在这种情况下，我们只处理一个文件，这样即使我们生成更多的任务（根据maxTask参数），我们返回一个列表，只有一个入口：		# 最后，是实现真正核心的 '''getTaskConfigs()'''。在这种情况下，我们只处理一个文件，这样即使我们生成更多的任务（根据maxTask参数），我们返回一个列表，只有一个入口：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	@Override		@Override
	public List<Map<String, String>> getTaskConfigs(int maxTasks) {		public List<Map<String, String>> getTaskConfigs(int maxTasks) {
第196行：		第196行：
	==== Task例子：Source Task ====		==== Task例子：Source Task ====
	# 和连接器一样，我们需要创建一个类（继承基于Task的类）。它也有一些标准的生命周期方法：		# 和连接器一样，我们需要创建一个类（继承基于Task的类）。它也有一些标准的生命周期方法：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	public class FileStreamSourceTask extends SourceTask<Object, Object> {		public class FileStreamSourceTask extends SourceTask<Object, Object> {
	String filename;		String filename;
第214行：		第214行：
	</syntaxhighlight>		</syntaxhighlight>
	接下来，我们实现任务的主要功能，poll()方法。它从输入系统获取时间并返回一个List<SourceRecord>：		接下来，我们实现任务的主要功能，poll()方法。它从输入系统获取时间并返回一个List<SourceRecord>：
	#: <syntaxhighlight lang="~~bash~~" highlight="">		#: <syntaxhighlight lang="java" highlight="">
	@Override		@Override
	public List<SourceRecord> poll() throws InterruptedException {		public List<SourceRecord> poll() throws InterruptedException {
第248行：		第248行：

	两者都有共同的生命周期的方法，但是SinkTask接口是完全不同的：		两者都有共同的生命周期的方法，但是SinkTask接口是完全不同的：
	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	public abstract class SinkTask implements Task {		public abstract class SinkTask implements Task {
	public void initialize(SinkTaskContext context) {		public void initialize(SinkTaskContext context) {
第266行：		第266行：
	SourceTask包含一个流ID（输入的文件名）和每个记录的offset（文件中的位置）。框架使用了定时提交offset，所以在故障的情况下，任务恢复并减少再处理和可能重复的事件数（如果Kafka Connect正常的停止，可从最近的offset恢复，例如在独立模式或重新加载配置）。提交处理是完全自动化的，但只有connector知道如何返回到正确的位置，从该位置恢复。正确的恢复后，任务可以使用SourceContext传递其initialize()方法来访问offset数据。在initialize()中，我们会添加一些代码来读取offset（如果存在），并找到它的位置。		SourceTask包含一个流ID（输入的文件名）和每个记录的offset（文件中的位置）。框架使用了定时提交offset，所以在故障的情况下，任务恢复并减少再处理和可能重复的事件数（如果Kafka Connect正常的停止，可从最近的offset恢复，例如在独立模式或重新加载配置）。提交处理是完全自动化的，但只有connector知道如何返回到正确的位置，从该位置恢复。正确的恢复后，任务可以使用SourceContext传递其initialize()方法来访问offset数据。在initialize()中，我们会添加一些代码来读取offset（如果存在），并找到它的位置。

	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	stream = new FileInputStream(filename);		stream = new FileInputStream(filename);
	Map<String, Object> offset = context.offsetStorageReader().offset(Collections.singletonMap(FILENAME_FIELD, filename));		Map<String, Object> offset = context.offsetStorageReader().offset(Collections.singletonMap(FILENAME_FIELD, filename));
第282行：		第282行：

	例如，在SourceConnector：		例如，在SourceConnector：
	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	if (inputsChanged())		if (inputsChanged())
	this.context.requestTaskReconfiguration();		this.context.requestTaskReconfiguration();
第293行：		第293行：

	下面的代码在 FileStreamSourceConnector 定义配置和暴露给框架。		下面的代码在 FileStreamSourceConnector 定义配置和暴露给框架。
	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	private static final ConfigDef CONFIG_DEF = new ConfigDef()		private static final ConfigDef CONFIG_DEF = new ConfigDef()
	.define(FILE_CONFIG, Type.STRING, Importance.HIGH, "Source filename.")		.define(FILE_CONFIG, Type.STRING, Importance.HIGH, "Source filename.")
第309行：		第309行：

	这里是一个简单的例子，创建一个Schema和Struct：		这里是一个简单的例子，创建一个Schema和Struct：
	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	Schema schema = SchemaBuilder.struct().name(NAME)		Schema schema = SchemaBuilder.struct().name(NAME)
	.field("name", Schema.STRING_SCHEMA)		.field("name", Schema.STRING_SCHEMA)
第332行：		第332行：

	例如，查询一个源文件的状态（使用 GET /connectors/file-source/status）可能会产生如下的输入：		例如，查询一个源文件的状态（使用 GET /connectors/file-source/status）可能会产生如下的输入：
	<syntaxhighlight lang="~~bash~~" highlight="">		<syntaxhighlight lang="java" highlight="">
	{		{
	"name": "file-source",		"name": "file-source",

Eijux：/* 开发指南 */

2021-05-20T04:26:08Z

开发指南

←上一版本		2021年5月20日 (四) 12:26的版本
第89行：		第89行：
	* '''PUT /connector-plugins/{connector-type}/config/validate'''：对提供的配置值进行验证，执行对每个配置验证，返回验证的建议值和错误信息。		* '''PUT /connector-plugins/{connector-type}/config/validate'''：对提供的配置值进行验证，执行对每个配置验证，返回验证的建议值和错误信息。

	== ~~开发指南~~ ==		== 开发指南【？？？】 ==
	开发者怎么样编写新的connector，用于kafka和其他系统之间的数据移动。		开发者怎么样编写新的connector，用于kafka和其他系统之间的数据移动。

2021年5月20日 (四) 04:25 Eijux

2021-05-20T04:25:44Z

显示更改

Eijux：建立内容为“category:Kafka == 关于 == ”的新页面

2021-05-20T03:26:53Z

建立内容为“category:Kafka == 关于 == <syntaxhighlight lang="java" highlight=""> </syntaxhighlight>”的新页面

新页面

[[category:Kafka]]

== 关于 ==

<syntaxhighlight lang="java" highlight="">

</syntaxhighlight>

Kafka：Connect - 版本历史

Eijux：​/* 开发指南【？？？】 */

Eijux：​/* 开发指南 */

2021年5月20日 (四) 04:25 Eijux

Eijux：​建立内容为“category:Kafka == 关于 == ”的新页面

Eijux：/* 开发指南【？？？】 */

Eijux：/* 开发指南 */

Eijux：建立内容为“category:Kafka == 关于 == ”的新页面