开心一刻

昨天发了一条朋友圈:酒吧有什么好去的,上个月在酒吧当服务员兼职,一位大姐看上了我,说一个月给我 10 万,要我陪她去上海,我没同意

朋友评论道:你没同意,为什么在上海?

我回复到:上个月没同意

嘴真硬

前情回顾

关于
DataX
,官网有很详细的介绍,鄙人不才,也写过几篇文章

异构数据源同步之数据同步 → datax 改造,有点意思

异构数据源同步之数据同步 → datax 再改造,开始触及源码

异构数据源同步之数据同步 → DataX 使用细节

异构数据源数据同步 → 从源码分析 DataX 敏感信息的加解密

不了解的小伙伴可以按需去查看,所以了,
DataX
就不做过多介绍了;官方提供了非常多的插件,囊括了绝大部分的数据源,基本可以满足我们日常需要,但数据源种类太多,DataX 插件不可能包含全部,比如
kafka
,DataX 官方是没有提供读写插件的,大家知道为什么吗?你们如果对数据同步了解的比较多的话,一看到 kafka,第一反应往往想到的是
实时同步
,而 DataX 针对的是
离线同步
,所以 DataX 官方没提供 kafka 插件是不是也就能理解了?因为不合适嘛!

但如果客户非要离线同步也支持 kafka

人家要嘛

你能怎么办?直接怼过去:实现不了?

实现不了

所以没得选,那就只能给 DataX 开发一套 kafka 插件了;基于
DataX插件开发宝典
,插件开发起来还是非常简单的

kafkawriter

  1. 编程接口

    自定义
    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<Configuration> split(int mandatoryNumber) {
                List<Configuration> configurations = new ArrayList<Configuration>(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<String, String> producer;
            private Configuration conf;
            private Properties props;
            private String fieldDelimiter;
            private List<String> 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<String, String>(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<NewTopic> newTopics = new ArrayList<NewTopic>();
                            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<String, String>(this.conf.getString(Key.TOPIC),
                                recordToString(record),
                                recordToString(record))
                        );
                    } else if (writeType.equalsIgnoreCase(WriteType.JSON.name())) {
                        producer.send(new ProducerRecord<String, String>(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<KafkaColumn> 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 的接口


    job_task 接口执行顺序

    重点看 Task 的接口实现


    • init:读取配置项,然后创建 Producer 实例

    • prepare:判断 Topic 是否存在,不存在则创建

    • startWrite:通过 RecordReceiver 从 Channel 获取 Record,然后写入 Topic

      支持两种写入格式:
      text

      json
      ,细节请看下文中的
      kafkawriter.md

    • destroy:关闭 Producer 实例


    实现不难,相信大家都能看懂

  2. 插件定义


    resources
    下新增
    plugin.json

    {
        "name": "kafkawriter",
        "class": "com.qsl.datax.plugin.writer.kafkawriter.KafkaWriter",
        "description": "write data to kafka",
        "developer": "qsl"
    }
    

    强调下
    class
    ,是
    KafkaWriter
    的全限定类名,如果你们没有完全拷贝我的,那么要改成你们自己的

  3. 配置文件


    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

  4. 打包发布

    可以参考官方的
    assembly
    配置,利用 assembly 来打包

至此,
kafkawriter
就算完成了

kafkareader

  1. 编程接口

    自定义
    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<Configuration> split(int adviceNumber) {
                List<Configuration> configurations = new ArrayList<>(adviceNumber);
                for (int i=0; i<adviceNumber; i++) {
                    configurations.add(this.originalConfig.clone());
                }
                return configurations;
            }
    
            private void validateParameter() {
                this.originalConfig.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaReaderErrorCode.REQUIRED_VALUE);
                this.originalConfig.getNecessaryValue(Key.TOPIC, KafkaReaderErrorCode.REQUIRED_VALUE);
            }
        }
    
        public static class Task extends Reader.Task {
    
            private static final Logger logger = LoggerFactory.getLogger(Task.class);
    
            private Consumer<String, String> consumer;
            private String topic;
            private Configuration conf;
            private int maxPollRecords;
            private String fieldDelimiter;
            private String readType;
            private List<Column.Type> 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<String> 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<String, String> 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<String, String> records) {
                if (CollUtil.isEmpty(records)) {
                    return;
                }
                for (ConsumerRecord<String, String> 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<String> 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<KafkaColumn> 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<columnTypes.size(); i++) {
                    KafkaColumn kafkaColumn = kafkaColumns.get(i);
                    switch (columnTypes.get(i)) {
                        case STRING:
                            sendRecord.setColumn(i, new StringColumn(kafkaColumn.getColumnValue()));
                            break;
                        case LONG:
                            sendRecord.setColumn(i, new LongColumn(kafkaColumn.getColumnValue()));
                            break;
                        case DOUBLE:
                            sendRecord.setColumn(i, new DoubleColumn(kafkaColumn.getColumnValue()));
                            break;
                        case DATE:
                            // 暂只支持时间戳
                            sendRecord.setColumn(i, new DateColumn(Long.parseLong(kafkaColumn.getColumnValue())));
                            break;
                        case BOOL:
                            sendRecord.setColumn(i, new BoolColumn(kafkaColumn.getColumnValue()));
                            break;
                        case BYTES:
                            sendRecord.setColumn(i, new BytesColumn(kafkaColumn.getColumnValue().getBytes(StandardCharsets.UTF_8)));
                            break;
                        default:
                            throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
                                    String.format("您提供的配置文件有误,datax不支持数据类型[%s]", columnTypes.get(i)));
                    }
                }
            }
        }
    }
    

    重点看 Task 的接口实现


    • init:读取配置项,然后创建 Consumer 实例

    • startWrite:从 Topic 拉取数据,通过 RecordSender 写入到 Channel 中

      这里有几个细节需要注意下


      1. Consumer 每次都是新创建的,拉取数据的时候,如果消费者还未加入到指定的消费者组中,那么它会先加入到消费者组中,加入过程会进行 Rebalance,而 Rebalance 会导致同一消费者组内的所有消费者都不能工作,此时即使 Topic 中有可拉取的消息,也拉取不到消息,所以引入了重试机制来尽量保证那一次同步任务拉取的时候,消费者能正常拉取消息
      2. 一旦 Consumer 拉取到消息,则会循环拉取消息,如果某一次的拉取数据量小于最大拉取量(maxPollRecords),说明 Topic 中的消息已经被拉取完了,那么循环终止;这与常规使用(Consumer 会一直主动拉取或被动接收)是有差别的
      3. 支持两种读取格式:
        text

        json
        ,细节请看下文的配置文件说明
      4. 为了保证写入 Channel 数据的完整,需要配置列的数据类型(DataX 的数据类型)
    • destroy:

      关闭 Consumer 实例

  2. 插件定义


    resources
    下新增
    plugin.json

    {
        "name": "kafkareader",
        "class": "com.qsl.datax.plugin.reader.kafkareader.KafkaReader",
        "description": "read data from kafka",
        "developer": "qsl"
    }
    

    class

    KafkaReader
    的全限定类名

  3. 配置文件


    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

  4. 打包发布

    可以参考官方的
    assembly
    配置,利用 assembly 来打包

至此,
kafkareader
也完成了

总结

  1. 完整代码:
    qsl-datax
  2. kafkareader 重试机制只能降低拉取不到数据的概率,并不能杜绝;另外,如果上游一直往 Topic 中发消息,kafkareader 每次拉取的数据量都等于最大拉取量,那么同步任务会一直进行而不会停止,这还是离线同步吗?
  3. 离线同步,不推荐走 kafka,因为用 kafka 走实时同步更香

标签: none

添加新评论