开心一刻 昨天发了一条朋友圈：酒吧有什么好去的，上个月在酒吧当服务员兼职，一位大姐看上了我，说一个月给我 10 万，要我陪她去上海，我没同意 朋友评论道：你没同意，为什么在上海？ 我回复到：上个月没同意 前情回顾 关于 DataX，官网有很详细的介绍，鄙人不才，也写过几篇文章 异构数据源同步之数据同
昨天发了一条朋友圈：酒吧有什么好去的，上个月在酒吧当服务员兼职，一位大姐看上了我，说一个月给我 10 万，要我陪她去上海，我没同意
朋友评论道：你没同意，为什么在上海？
我回复到：上个月没同意
关于 DataX ，官网有很详细的介绍，鄙人不才，也写过几篇文章
异构数据源同步之数据同步 → datax 改造，有点意思
异构数据源同步之数据同步 → datax 再改造，开始触及源码
异构数据源同步之数据同步 → DataX 使用细节
异构数据源数据同步 → 从源码分析 DataX 敏感信息的加解密
不了解的小伙伴可以按需去查看，所以了，
DataX
就不做过多介绍了；官方提供了非常多的插件，囊括了绝大部分的数据源，基本可以满足我们日常需要，但数据源种类太多，DataX 插件不可能包含全部，比如
kafka
，DataX 官方是没有提供读写插件的，大家知道为什么吗？你们如果对数据同步了解的比较多的话，一看到 kafka，第一反应往往想到的是
实时同步
，而 DataX 针对的是
离线同步
，所以 DataX 官方没提供 kafka 插件是不是也就能理解了？因为不合适嘛！
但如果客户非要离线同步也支持 kafka
你能怎么办？直接怼过去：实现不了？
所以没得选，那就只能给 DataX 开发一套 kafka 插件了；基于 DataX插件开发宝典 ，插件开发起来还是非常简单的
编程接口
自定义
Kafkawriter
继承 DataX 的
Writer
，实现 job、task 对应的接口即可
/**
* @author 青石路
*/
public class KafkaWriter extends Writer {
public static class Job extends Writer.Job {
private Configuration conf = null;
@Override
public List split(int mandatoryNumber) {
List configurations = new ArrayList(mandatoryNumber);
for (int i = 0; i < mandatoryNumber; i++) {
configurations.add(this.conf.clone());
}
return configurations;
}
private void validateParameter() {
this.conf.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaWriterErrorCode.REQUIRED_VALUE);
this.conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
}
@Override
public void init() {
this.conf = super.getPluginJobConf();
this.validateParameter();
}
@Override
public void destroy() {
}
}
public static class Task extends Writer.Task {
private static final Logger logger = LoggerFactory.getLogger(Task.class);
private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");
private Producer producer;
private Configuration conf;
private Properties props;
private String fieldDelimiter;
private List columns;
private String writeType;
@Override
public void init() {
this.conf = super.getPluginJobConf();
fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
columns = conf.getList(Key.COLUMN, String.class);
writeType = conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null);
if (CollUtil.isEmpty(columns)) {
throw DataXException.asDataXException(KafkaWriterErrorCode.REQUIRED_VALUE,
String.format("您提供配置文件有误，[%s]是必填参数，不允许为空或者留白 .", Key.COLUMN));
}
props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, conf.getString(Key.BOOTSTRAP_SERVERS));
//这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
props.put(ProducerConfig.ACKS_CONFIG, conf.getUnnecessaryValue(Key.ACK, "0", null));
props.put(CommonClientConfigs.RETRIES_CONFIG, conf.getUnnecessaryValue(Key.RETRIES, "0", null));
props.put(ProducerConfig.BATCH_SIZE_CONFIG, conf.getUnnecessaryValue(Key.BATCH_SIZE, "16384", null));
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
Configuration saslConf = conf.getConfiguration(Key.SASL);
if (ObjUtil.isNotNull(saslConf)) {
logger.info("配置启用了SASL认证");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, saslConf.getNecessaryValue(Key.SASL_SECURITY_PROTOCOL, KafkaWriterErrorCode.REQUIRED_VALUE));
props.put(SaslConfigs.SASL_MECHANISM, saslConf.getNecessaryValue(Key.SASL_MECHANISM, KafkaWriterErrorCode.REQUIRED_VALUE));
String userName = saslConf.getNecessaryValue(Key.SASL_USERNAME, KafkaWriterErrorCode.REQUIRED_VALUE);
String password = saslConf.getNecessaryValue(Key.SASL_PASSWORD, KafkaWriterErrorCode.REQUIRED_VALUE);
props.put(SaslConfigs.SASL_JAAS_CONFIG, String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", userName, password));
}
producer = new KafkaProducer(props);
}
@Override
public void prepare() {
if (Boolean.parseBoolean(conf.getUnnecessaryValue(Key.NO_TOPIC_CREATE, "false", null))) {
ListTopicsResult topicsResult = AdminClient.create(props).listTopics();
String topic = conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
try {
if (!topicsResult.names().get().contains(topic)) {
new NewTopic(
topic,
Integer.parseInt(conf.getUnnecessaryValue(Key.TOPIC_NUM_PARTITION, "1", null)),
Short.parseShort(conf.getUnnecessaryValue(Key.TOPIC_REPLICATION_FACTOR, "1", null))
);
List newTopics = new ArrayList();
AdminClient.create(props).createTopics(newTopics);
}
} catch (Exception e) {
throw new DataXException(KafkaWriterErrorCode.CREATE_TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE.getDescription());
}
}
}
@Override
public void startWrite(RecordReceiver lineReceiver) {
logger.info("start to writer kafka");
Record record = null;
while ((record = lineReceiver.getFromReader()) != null) {//说明还在读取数据,或者读取的数据没处理完
//获取一行数据，按照指定分隔符 拼成字符串 发送出去
if (writeType.equalsIgnoreCase(WriteType.TEXT.name())) {
producer.send(new ProducerRecord(this.conf.getString(Key.TOPIC),
recordToString(record),
recordToString(record))
);
} else if (writeType.equalsIgnoreCase(WriteType.JSON.name())) {
producer.send(new ProducerRecord(this.conf.getString(Key.TOPIC),
recordToString(record),
recordToKafkaJson(record))
);
}
producer.flush();
}
}
@Override
public void destroy() {
logger.info("producer close");
if (producer != null) {
producer.close();
}
}
/**
* 数据格式化
*
* @param record
* @return
*/
private String recordToString(Record record) {
int recordLength = record.getColumnNumber();
if (0 == recordLength) {
return NEWLINE_FLAG;
}
Column column;
StringBuilder sb = new StringBuilder();
for (int i = 0; i < recordLength; i++) {
column = record.getColumn(i);
sb.append(column.asString()).append(fieldDelimiter);
}
sb.setLength(sb.length() - 1);
sb.append(NEWLINE_FLAG);
return sb.toString();
}
private String recordToKafkaJson(Record record) {
int recordLength = record.getColumnNumber();
if (recordLength != columns.size()) {
throw DataXException.asDataXException(KafkaWriterErrorCode.ILLEGAL_PARAM,
String.format("您提供配置文件有误，列数不匹配[record columns=%d, writer columns=%d]", recordLength, columns.size()));
}
List kafkaColumns = new ArrayList<>();
for (int i = 0; i < recordLength; i++) {
KafkaColumn column = new KafkaColumn(record.getColumn(i), columns.get(i));
kafkaColumns.add(column);
}
return JSONUtil.toJsonStr(kafkaColumns);
}
}
}
DataX 框架按照如下的顺序执行 Job 和 Task 的接口
重点看 Task 的接口实现
init：读取配置项，然后创建 Producer 实例
prepare：判断 Topic 是否存在，不存在则创建
startWrite：通过 RecordReceiver 从 Channel 获取 Record，然后写入 Topic
支持两种写入格式：
text
、
json
，细节请看下文中的
kafkawriter.md
destroy：关闭 Producer 实例
实现不难，相信大家都能看懂
插件定义
在
resources
下新增
plugin.json
{
"name": "kafkawriter",
"class": "com.qsl.datax.plugin.writer.kafkawriter.KafkaWriter",
"description": "write data to kafka",
"developer": "qsl"
}
强调下
class
，是
KafkaWriter
的全限定类名，如果你们没有完全拷贝我的，那么要改成你们自己的
配置文件
在
resources
下新增
plugin_job_template.json
{
"name": "kafkawriter",
"parameter": {
"bootstrapServers": "",
"topic": "",
"ack": "all",
"batchSize": 1000,
"retries": 0,
"fieldDelimiter": ",",
"writeType": "json",
"column": [
"const_id",
"const_field",
"const_field_value"
],
"sasl": {
"securityProtocol": "SASL_PLAINTEXT",
"mechanism": "PLAIN",
"username": "",
"password": ""
}
}
}
配置项说明： kafkawriter.md
打包发布
可以参考官方的
assembly
配置，利用 assembly 来打包
至此，
kafkawriter
就算完成了
编程接口
自定义
Kafkareader
继承 DataX 的
Reader
，实现 job、task 对应的接口即可
/**
* @author 青石路
*/
public class KafkaReader extends Reader {
public static class Job extends Reader.Job {
private Configuration originalConfig = null;
@Override
public void init() {
this.originalConfig = super.getPluginJobConf();
this.validateParameter();
}
@Override
public void destroy() {
}
@Override
public List split(int adviceNumber) {
List configurations = new ArrayList<>(adviceNumber);
for (int i=0; i consumer;
private String topic;
private Configuration conf;
private int maxPollRecords;
private String fieldDelimiter;
private String readType;
private List columnTypes;
@Override
public void destroy() {
logger.info("consumer close");
if (Objects.nonNull(consumer)) {
consumer.close();
}
}
@Override
public void init() {
this.conf = super.getPluginJobConf();
this.topic = conf.getString(Key.TOPIC);
this.maxPollRecords = conf.getInt(Key.MAX_POLL_RECORDS, 500);
fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
readType = conf.getUnnecessaryValue(Key.READ_TYPE, ReadType.JSON.name(), null);
if (!ReadType.JSON.name().equalsIgnoreCase(readType)
&& !ReadType.TEXT.name().equalsIgnoreCase(readType)) {
throw DataXException.asDataXException(KafkaReaderErrorCode.REQUIRED_VALUE,
String.format("您提供配置文件有误，不支持的readType[%s]", readType));
}
if (ReadType.JSON.name().equalsIgnoreCase(readType)) {
List columnTypeList = conf.getList(Key.COLUMN_TYPE, String.class);
if (CollUtil.isEmpty(columnTypeList)) {
throw DataXException.asDataXException(KafkaReaderErrorCode.REQUIRED_VALUE,
String.format("您提供配置文件有误，readType是JSON时[%s]是必填参数，不允许为空或者留白 .", Key.COLUMN_TYPE));
}
convertColumnType(columnTypeList);
}
Properties props = new Properties();
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, conf.getString(Key.BOOTSTRAP_SERVERS));
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer", null));
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer", null));
props.put(ConsumerConfig.GROUP_ID_CONFIG, conf.getNecessaryValue(Key.GROUP_ID, KafkaReaderErrorCode.REQUIRED_VALUE));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
Configuration saslConf = conf.getConfiguration(Key.SASL);
if (ObjUtil.isNotNull(saslConf)) {
logger.info("配置启用了SASL认证");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, saslConf.getNecessaryValue(Key.SASL_SECURITY_PROTOCOL, KafkaReaderErrorCode.REQUIRED_VALUE));
props.put(SaslConfigs.SASL_MECHANISM, saslConf.getNecessaryValue(Key.SASL_MECHANISM, KafkaReaderErrorCode.REQUIRED_VALUE));
String userName = saslConf.getNecessaryValue(Key.SASL_USERNAME, KafkaReaderErrorCode.REQUIRED_VALUE);
String password = saslConf.getNecessaryValue(Key.SASL_PASSWORD, KafkaReaderErrorCode.REQUIRED_VALUE);
props.put(SaslConfigs.SASL_JAAS_CONFIG, String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", userName, password));
}
consumer = new KafkaConsumer<>(props);
}
@Override
public void startRead(RecordSender recordSender) {
consumer.subscribe(CollUtil.newArrayList(topic));
int pollTimeoutMs = conf.getInt(Key.POLL_TIMEOUT_MS, 1000);
int retries = conf.getInt(Key.RETRIES, 5);
if (retries < 0) {
logger.info("joinGroupSuccessRetries 配置有误[{}], 重置成默认值[5]", retries);
retries = 5;
}
/**
* consumer 每次都是新创建，第一次poll时会重新加入消费者组，加入过程会进行Rebalance，而 Rebalance 会导致同一 Group 内的所有消费者都不能工作
* 所以 poll 拉取的过程中，即使topic中有数据也不一定能拉到，因为 consumer 正在加入消费者组中
* kafka-clients 没有对应的API、事件机制来知道 consumer 成功加入消费者组的确切时间
* 故增加重试
*/
ConsumerRecords records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
int i = 0;
if (CollUtil.isEmpty(records)) {
for (; i < retries; i++) {
records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
logger.info("第 {} 次重试，获取消息记录数[{}]", i + 1, records.count());
if (!CollUtil.isEmpty(records)) {
break;
}
}
}
if (i >= retries) {
logger.info("重试 {} 次后，仍未获取到消息，请确认是否有数据、配置是否正确", retries);
return;
}
transferRecord(recordSender, records);
do {
records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
transferRecord(recordSender, records);
} while (!CollUtil.isEmpty(records) && records.count() >= maxPollRecords);
}
private void transferRecord(RecordSender recordSender, ConsumerRecords records) {
if (CollUtil.isEmpty(records)) {
return;
}
for (ConsumerRecord record : records) {
Record sendRecord = recordSender.createRecord();
String msgValue = record.value();
if (ReadType.JSON.name().equalsIgnoreCase(readType)) {
transportJsonToRecord(sendRecord, msgValue);
} else if (ReadType.TEXT.name().equalsIgnoreCase(readType)) {
// readType = text，全当字符串类型处理
String[] columnValues = msgValue.split(fieldDelimiter);
for (String columnValue : columnValues) {
sendRecord.addColumn(new StringColumn(columnValue));
}
}
recordSender.sendToWriter(sendRecord);
}
consumer.commitAsync();
}
private void convertColumnType(List columnTypeList) {
columnTypes = new ArrayList<>();
for (String columnType : columnTypeList) {
switch (columnType.toUpperCase()) {
case "STRING":
columnTypes.add(Column.Type.STRING);
break;
case "LONG":
columnTypes.add(Column.Type.LONG);
break;
case "DOUBLE":
columnTypes.add(Column.Type.DOUBLE);
case "DATE":
columnTypes.add(Column.Type.DATE);
break;
case "BOOLEAN":
columnTypes.add(Column.Type.BOOL);
break;
case "BYTES":
columnTypes.add(Column.Type.BYTES);
break;
default:
throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
String.format("您提供的配置文件有误，datax不支持数据类型[%s]", columnType));
}
}
}
private void transportJsonToRecord(Record sendRecord, String msgValue) {
List kafkaColumns = JSONUtil.toList(msgValue, KafkaColumn.class);
if (columnTypes.size() != kafkaColumns.size()) {
throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
String.format("您提供的配置文件有误，readType是JSON时[%s列数=%d]与[json列数=%d]的数量不匹配", Key.COLUMN_TYPE, columnTypes.size(), kafkaColumns.size()));
}
for (int i=0; i 重点看 Task 的接口实现
init：读取配置项，然后创建 Consumer 实例
startWrite：从 Topic 拉取数据，通过 RecordSender 写入到 Channel 中
这里有几个细节需要注意下
text
、
json
，细节请看下文的配置文件说明
destroy：
关闭 Consumer 实例
插件定义
在
resources
下新增
plugin.json
{
"name": "kafkareader",
"class": "com.qsl.datax.plugin.reader.kafkareader.KafkaReader",
"description": "read data from kafka",
"developer": "qsl"
}
class
是
KafkaReader
的全限定类名
配置文件
在
resources
下新增
plugin_job_template.json
{
"name": "kafkareader",
"parameter": {
"bootstrapServers": "",
"topic": "test-kafka",
"groupId": "test1",
"writeType": "json",
"pollTimeoutMs": 2000,
"columnType": [
"LONG",
"STRING",
"STRING"
],
"sasl": {
"securityProtocol": "SASL_PLAINTEXT",
"mechanism": "PLAIN",
"username": "",
"password": "2"
}
}
}
配置项说明： kafkareader.md
打包发布
可以参考官方的
assembly
配置，利用 assembly 来打包
至此，
kafkareader
也完成了
