我有一个用例,我需要编写自定义逻辑来根据消息中的某些关键参数分配分区。我对此做了一些研究,发现卡夫卡转换支持覆盖转换接口中的某些方法,但我无法在 git hub 或其他地方执行一些示例代码。有人可以分享示例代码或 git hub 链接来在 kafka JDBC 源连接器中进行自定义分区分配吗?
提前致谢!。
卡夫卡连接默认情况下分配分区使用:DefaultPartitioner
(org.apache.kafka.clients.producer.internals.DefaultPartitioner
)
如果您需要使用某些自定义覆盖默认值,这是可能的,但您必须记住,覆盖适用于所有源连接器。
为此,您必须设置producer.partitioner.class
财产,前producer.partitioner.class=com.example.CustomPartitioner
。
此外,您必须使用分区器将 jar 复制到 Kafka Connect 库的目录中。
改造方式:
在 Transformation 中也可以设置分区,但这不是正确的方法。
从Transformation
您无权访问主题元数据,这对于分配分区至关重要。
无论如何,如果您想为记录设置分区,代码应如下所示:
public class AddPartition <R extends ConnectRecord<R>> implements Transformation<R> {
public static final ConfigDef CONFIG_DEF = new ConfigDef();
@Override
public void configure(Map<String, ?> props) {
final SimpleConfig config = new SimpleConfig(CONFIG_DEF, props);
}
@Override
public R apply(R record) {
return record.newRecord(record.topic(), calculatePartition(record), record.keySchema(), record.key(), record.valueSchema(), record.value(), record.timestamp());
}
private Integer calculatePartition(R record) {
// Partitions calcuation based on record information
return 0;
}
@Override
public void close() {
}
@Override
public ConfigDef config() {
return CONFIG_DEF;
}
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)