经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Apache Kafka » 查看文章
异源数据同步 → DataX 为什么要支持 kafka？
来源：cnblogs　　作者：青石路　　时间：2024/8/26 9:16:15　　对本文有异议

开心一刻

昨天发了一条朋友圈：酒吧有什么好去的，上个月在酒吧当服务员兼职，一位大姐看上了我，说一个月给我 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 对应的接口即可

    1. /**
    2.  * @author 青石路
    3.  */
    4. public class KafkaWriter extends Writer {
    6.     public static class Job extends Writer.Job {
    8.         private Configuration conf = null;
    10.         @Override
    11.         public List<Configuration> split(int mandatoryNumber) {
    12.             List<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber);
    13.             for (int i = 0; i < mandatoryNumber; i++) {
    14.                 configurations.add(this.conf.clone());
    15.             }
    16.             return configurations;
    17.         }
    19.         private void validateParameter() {
    20.             this.conf.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaWriterErrorCode.REQUIRED_VALUE);
    21.             this.conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
    22.         }
    24.         @Override
    25.         public void init() {
    26.             this.conf = super.getPluginJobConf();
    27.             this.validateParameter();
    28.         }
    31.         @Override
    32.         public void destroy() {
    34.         }
    35.     }
    37.     public static class Task extends Writer.Task {
    38.         private static final Logger logger = LoggerFactory.getLogger(Task.class);
    39.         private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");
    41.         private Producer<String, String> producer;
    42.         private Configuration conf;
    43.         private Properties props;
    44.         private String fieldDelimiter;
    45.         private List<String> columns;
    46.         private String writeType;
    48.         @Override
    49.         public void init() {
    50.             this.conf = super.getPluginJobConf();
    51.             fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
    52.             columns = conf.getList(Key.COLUMN, String.class);
    53.             writeType = conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null);
    54.             if (CollUtil.isEmpty(columns)) {
    55.                 throw DataXException.asDataXException(KafkaWriterErrorCode.REQUIRED_VALUE,
    56.                         String.format("您提供配置文件有误，[%s]是必填参数，不允许为空或者留白 .", Key.COLUMN));
    57.             }
    59.             props = new Properties();
    60.             props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, conf.getString(Key.BOOTSTRAP_SERVERS));
    61.             //这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
    62.             props.put(ProducerConfig.ACKS_CONFIG, conf.getUnnecessaryValue(Key.ACK, "0", null));
    63.             props.put(CommonClientConfigs.RETRIES_CONFIG, conf.getUnnecessaryValue(Key.RETRIES, "0", null));
    64.             props.put(ProducerConfig.BATCH_SIZE_CONFIG, conf.getUnnecessaryValue(Key.BATCH_SIZE, "16384", null));
    65.             props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    66.             props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
    67.             props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
    69.             Configuration saslConf = conf.getConfiguration(Key.SASL);
    70.             if (ObjUtil.isNotNull(saslConf)) {
    71.                 logger.info("配置启用了SASL认证");
    72.                 props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, saslConf.getNecessaryValue(Key.SASL_SECURITY_PROTOCOL, KafkaWriterErrorCode.REQUIRED_VALUE));
    73.                 props.put(SaslConfigs.SASL_MECHANISM, saslConf.getNecessaryValue(Key.SASL_MECHANISM, KafkaWriterErrorCode.REQUIRED_VALUE));
    74.                 String userName = saslConf.getNecessaryValue(Key.SASL_USERNAME, KafkaWriterErrorCode.REQUIRED_VALUE);
    75.                 String password = saslConf.getNecessaryValue(Key.SASL_PASSWORD, KafkaWriterErrorCode.REQUIRED_VALUE);
    76.                 props.put(SaslConfigs.SASL_JAAS_CONFIG, String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", userName, password));
    77.             }
    79.             producer = new KafkaProducer<String, String>(props);
    80.         }
    82.         @Override
    83.         public void prepare() {
    84.             if (Boolean.parseBoolean(conf.getUnnecessaryValue(Key.NO_TOPIC_CREATE, "false", null))) {
    86.                 ListTopicsResult topicsResult = AdminClient.create(props).listTopics();
    87.                 String topic = conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
    89.                 try {
    90.                     if (!topicsResult.names().get().contains(topic)) {
    91.                         new NewTopic(
    92.                                 topic,
    93.                                 Integer.parseInt(conf.getUnnecessaryValue(Key.TOPIC_NUM_PARTITION, "1", null)),
    94.                                 Short.parseShort(conf.getUnnecessaryValue(Key.TOPIC_REPLICATION_FACTOR, "1", null))
    95.                         );
    96.                         List<NewTopic> newTopics = new ArrayList<NewTopic>();
    97.                         AdminClient.create(props).createTopics(newTopics);
    98.                     }
    99.                 } catch (Exception e) {
    100.                     throw new DataXException(KafkaWriterErrorCode.CREATE_TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE.getDescription());
    101.                 }
    102.             }
    103.         }
    105.         @Override
    106.         public void startWrite(RecordReceiver lineReceiver) {
    107.             logger.info("start to writer kafka");
    108.             Record record = null;
    109.             while ((record = lineReceiver.getFromReader()) != null) {//说明还在读取数据,或者读取的数据没处理完
    110.                 //获取一行数据，按照指定分隔符 拼成字符串 发送出去
    111.                 if (writeType.equalsIgnoreCase(WriteType.TEXT.name())) {
    112.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
    113.                             recordToString(record),
    114.                             recordToString(record))
    115.                     );
    116.                 } else if (writeType.equalsIgnoreCase(WriteType.JSON.name())) {
    117.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
    118.                             recordToString(record),
    119.                             recordToKafkaJson(record))
    120.                     );
    121.                 }
    122.                 producer.flush();
    123.             }
    124.         }
    126.         @Override
    127.         public void destroy() {
    128.             logger.info("producer close");
    129.             if (producer != null) {
    130.                 producer.close();
    131.             }
    132.         }
    134.         /**
    135.          * 数据格式化
    136.          *
    137.          * @param record
    138.          * @return
    139.          */
    140.         private String recordToString(Record record) {
    141.             int recordLength = record.getColumnNumber();
    142.             if (0 == recordLength) {
    143.                 return NEWLINE_FLAG;
    144.             }
    145.             Column column;
    146.             StringBuilder sb = new StringBuilder();
    147.             for (int i = 0; i < recordLength; i++) {
    148.                 column = record.getColumn(i);
    149.                 sb.append(column.asString()).append(fieldDelimiter);
    150.             }
    152.             sb.setLength(sb.length() - 1);
    153.             sb.append(NEWLINE_FLAG);
    155.             return sb.toString();
    156.         }
    158.         private String recordToKafkaJson(Record record) {
    159.             int recordLength = record.getColumnNumber();
    160.             if (recordLength != columns.size()) {
    161.                 throw DataXException.asDataXException(KafkaWriterErrorCode.ILLEGAL_PARAM,
    162.                         String.format("您提供配置文件有误，列数不匹配[record columns=%d, writer columns=%d]", recordLength, columns.size()));
    163.             }
    164.             List<KafkaColumn> kafkaColumns = new ArrayList<>();
    165.             for (int i = 0; i < recordLength; i++) {
    166.                 KafkaColumn column = new KafkaColumn(record.getColumn(i), columns.get(i));
    167.                 kafkaColumns.add(column);
    168.             }
    169.             return JSONUtil.toJsonStr(kafkaColumns);
    170.         }
    171.     }
    172. }

    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

    1. {
    2.     "name": "kafkawriter",
    3.     "class": "com.qsl.datax.plugin.writer.kafkawriter.KafkaWriter",
    4.     "description": "write data to kafka",
    5.     "developer": "qsl"
    6. }

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

  3. 配置文件

    在 resources 下新增 plugin_job_template.json

    1. {
    2.     "name": "kafkawriter",
    3.     "parameter": {
    4.         "bootstrapServers": "",
    5.         "topic": "",
    6.         "ack": "all",
    7.         "batchSize": 1000,
    8.         "retries": 0,
    9.         "fieldDelimiter": ",",
    10.         "writeType": "json",
    11.         "column": [
    12.             "const_id",
    13.             "const_field",
    14.             "const_field_value"
    15.         ],
    16.         "sasl": {
    17.             "securityProtocol": "SASL_PLAINTEXT",
    18.             "mechanism": "PLAIN",
    19.             "username": "",
    20.             "password": ""
    21.         }
    22.     }
    23. }

    配置项说明：kafkawriter.md

  4. 打包发布

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

至此，kafkawriter 就算完成了

kafkareader

  1. 编程接口

    自定义 Kafkareader 继承 DataX 的 Reader，实现 job、task 对应的接口即可

    1. /**
    2.  * @author 青石路
    3.  */
    4. public class KafkaReader extends Reader {
    6.     public static class Job extends Reader.Job {
    8.         private Configuration originalConfig = null;
    10.         @Override
    11.         public void init() {
    12.             this.originalConfig = super.getPluginJobConf();
    13.             this.validateParameter();
    14.         }
    16.         @Override
    17.         public void destroy() {
    19.         }
    21.         @Override
    22.         public List<Configuration> split(int adviceNumber) {
    23.             List<Configuration> configurations = new ArrayList<>(adviceNumber);
    24.             for (int i=0; i<adviceNumber; i++) {
    25.                 configurations.add(this.originalConfig.clone());
    26.             }
    27.             return configurations;
    28.         }
    30.         private void validateParameter() {
    31.             this.originalConfig.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaReaderErrorCode.REQUIRED_VALUE);
    32.             this.originalConfig.getNecessaryValue(Key.TOPIC, KafkaReaderErrorCode.REQUIRED_VALUE);
    33.         }
    34.     }
    36.     public static class Task extends Reader.Task {
    38.         private static final Logger logger = LoggerFactory.getLogger(Task.class);
    40.         private Consumer<String, String> consumer;
    41.         private String topic;
    42.         private Configuration conf;
    43.         private int maxPollRecords;
    44.         private String fieldDelimiter;
    45.         private String readType;
    46.         private List<Column.Type> columnTypes;
    48.         @Override
    49.         public void destroy() {
    50.             logger.info("consumer close");
    51.             if (Objects.nonNull(consumer)) {
    52.                 consumer.close();
    53.             }
    54.         }
    56.         @Override
    57.         public void init() {
    58.             this.conf = super.getPluginJobConf();
    59.             this.topic = conf.getString(Key.TOPIC);
    60.             this.maxPollRecords = conf.getInt(Key.MAX_POLL_RECORDS, 500);
    61.             fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
    62.             readType = conf.getUnnecessaryValue(Key.READ_TYPE, ReadType.JSON.name(), null);
    63.             if (!ReadType.JSON.name().equalsIgnoreCase(readType)
    64.                     && !ReadType.TEXT.name().equalsIgnoreCase(readType)) {
    65.                 throw DataXException.asDataXException(KafkaReaderErrorCode.REQUIRED_VALUE,
    66.                         String.format("您提供配置文件有误，不支持的readType[%s]", readType));
    67.             }
    68.             if (ReadType.JSON.name().equalsIgnoreCase(readType)) {
    69.                 List<String> columnTypeList = conf.getList(Key.COLUMN_TYPE, String.class);
    70.                 if (CollUtil.isEmpty(columnTypeList)) {
    71.                     throw DataXException.asDataXException(KafkaReaderErrorCode.REQUIRED_VALUE,
    72.                             String.format("您提供配置文件有误，readType是JSON时[%s]是必填参数，不允许为空或者留白 .", Key.COLUMN_TYPE));
    73.                 }
    74.                 convertColumnType(columnTypeList);
    75.             }
    76.             Properties props = new Properties();
    77.             props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, conf.getString(Key.BOOTSTRAP_SERVERS));
    78.             props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer", null));
    79.             props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, conf.getUnnecessaryValue(Key.VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer", null));
    80.             props.put(ConsumerConfig.GROUP_ID_CONFIG, conf.getNecessaryValue(Key.GROUP_ID, KafkaReaderErrorCode.REQUIRED_VALUE));
    81.             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    82.             props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    83.             props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
    84.             Configuration saslConf = conf.getConfiguration(Key.SASL);
    85.             if (ObjUtil.isNotNull(saslConf)) {
    86.                 logger.info("配置启用了SASL认证");
    87.                 props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, saslConf.getNecessaryValue(Key.SASL_SECURITY_PROTOCOL, KafkaReaderErrorCode.REQUIRED_VALUE));
    88.                 props.put(SaslConfigs.SASL_MECHANISM, saslConf.getNecessaryValue(Key.SASL_MECHANISM, KafkaReaderErrorCode.REQUIRED_VALUE));
    89.                 String userName = saslConf.getNecessaryValue(Key.SASL_USERNAME, KafkaReaderErrorCode.REQUIRED_VALUE);
    90.                 String password = saslConf.getNecessaryValue(Key.SASL_PASSWORD, KafkaReaderErrorCode.REQUIRED_VALUE);
    91.                 props.put(SaslConfigs.SASL_JAAS_CONFIG, String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", userName, password));
    92.             }
    93.             consumer = new KafkaConsumer<>(props);
    94.         }
    96.         @Override
    97.         public void startRead(RecordSender recordSender) {
    98.             consumer.subscribe(CollUtil.newArrayList(topic));
    99.             int pollTimeoutMs = conf.getInt(Key.POLL_TIMEOUT_MS, 1000);
    100.             int retries = conf.getInt(Key.RETRIES, 5);
    101.             if (retries < 0) {
    102.                 logger.info("joinGroupSuccessRetries 配置有误[{}], 重置成默认值[5]", retries);
    103.                 retries = 5;
    104.             }
    105.             /**
    106.              * consumer 每次都是新创建，第一次poll时会重新加入消费者组，加入过程会进行Rebalance，而 Rebalance 会导致同一 Group 内的所有消费者都不能工作
    107.              * 所以 poll 拉取的过程中，即使topic中有数据也不一定能拉到，因为 consumer 正在加入消费者组中
    108.              * kafka-clients 没有对应的API、事件机制来知道 consumer 成功加入消费者组的确切时间
    109.              * 故增加重试
    110.              */
    111.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
    112.             int i = 0;
    113.             if (CollUtil.isEmpty(records)) {
    114.                 for (; i < retries; i++) {
    115.                     records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
    116.                     logger.info("第 {} 次重试，获取消息记录数[{}]", i + 1, records.count());
    117.                     if (!CollUtil.isEmpty(records)) {
    118.                         break;
    119.                     }
    120.                 }
    121.             }
    122.             if (i >= retries) {
    123.                 logger.info("重试 {} 次后，仍未获取到消息，请确认是否有数据、配置是否正确", retries);
    124.                 return;
    125.             }
    126.             transferRecord(recordSender, records);
    127.             do {
    128.                 records = consumer.poll(Duration.ofMillis(pollTimeoutMs));
    129.                 transferRecord(recordSender, records);
    130.             } while (!CollUtil.isEmpty(records) && records.count() >= maxPollRecords);
    131.         }
    133.         private void transferRecord(RecordSender recordSender, ConsumerRecords<String, String> records) {
    134.             if (CollUtil.isEmpty(records)) {
    135.                 return;
    136.             }
    137.             for (ConsumerRecord<String, String> record : records) {
    138.                 Record sendRecord = recordSender.createRecord();
    139.                 String msgValue = record.value();
    140.                 if (ReadType.JSON.name().equalsIgnoreCase(readType)) {
    141.                     transportJsonToRecord(sendRecord, msgValue);
    142.                 } else if (ReadType.TEXT.name().equalsIgnoreCase(readType)) {
    143.                     // readType = text，全当字符串类型处理
    144.                     String[] columnValues = msgValue.split(fieldDelimiter);
    145.                     for (String columnValue : columnValues) {
    146.                         sendRecord.addColumn(new StringColumn(columnValue));
    147.                     }
    148.                 }
    149.                 recordSender.sendToWriter(sendRecord);
    150.             }
    151.             consumer.commitAsync();
    152.         }
    154.         private void convertColumnType(List<String> columnTypeList) {
    155.             columnTypes = new ArrayList<>();
    156.             for (String columnType : columnTypeList) {
    157.                 switch (columnType.toUpperCase()) {
    158.                     case "STRING":
    159.                         columnTypes.add(Column.Type.STRING);
    160.                         break;
    161.                     case "LONG":
    162.                         columnTypes.add(Column.Type.LONG);
    163.                         break;
    164.                     case "DOUBLE":
    165.                         columnTypes.add(Column.Type.DOUBLE);
    166.                     case "DATE":
    167.                         columnTypes.add(Column.Type.DATE);
    168.                         break;
    169.                     case "BOOLEAN":
    170.                         columnTypes.add(Column.Type.BOOL);
    171.                         break;
    172.                     case "BYTES":
    173.                         columnTypes.add(Column.Type.BYTES);
    174.                         break;
    175.                     default:
    176.                         throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
    177.                                 String.format("您提供的配置文件有误，datax不支持数据类型[%s]", columnType));
    178.                 }
    179.             }
    180.         }
    182.         private void transportJsonToRecord(Record sendRecord, String msgValue) {
    183.             List<KafkaColumn> kafkaColumns = JSONUtil.toList(msgValue, KafkaColumn.class);
    184.             if (columnTypes.size() != kafkaColumns.size()) {
    185.                 throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
    186.                         String.format("您提供的配置文件有误，readType是JSON时[%s列数=%d]与[json列数=%d]的数量不匹配", Key.COLUMN_TYPE, columnTypes.size(), kafkaColumns.size()));
    187.             }
    188.             for (int i=0; i<columnTypes.size(); i++) {
    189.                 KafkaColumn kafkaColumn = kafkaColumns.get(i);
    190.                 switch (columnTypes.get(i)) {
    191.                     case STRING:
    192.                         sendRecord.setColumn(i, new StringColumn(kafkaColumn.getColumnValue()));
    193.                         break;
    194.                     case LONG:
    195.                         sendRecord.setColumn(i, new LongColumn(kafkaColumn.getColumnValue()));
    196.                         break;
    197.                     case DOUBLE:
    198.                         sendRecord.setColumn(i, new DoubleColumn(kafkaColumn.getColumnValue()));
    199.                         break;
    200.                     case DATE:
    201.                         // 暂只支持时间戳
    202.                         sendRecord.setColumn(i, new DateColumn(Long.parseLong(kafkaColumn.getColumnValue())));
    203.                         break;
    204.                     case BOOL:
    205.                         sendRecord.setColumn(i, new BoolColumn(kafkaColumn.getColumnValue()));
    206.                         break;
    207.                     case BYTES:
    208.                         sendRecord.setColumn(i, new BytesColumn(kafkaColumn.getColumnValue().getBytes(StandardCharsets.UTF_8)));
    209.                         break;
    210.                     default:
    211.                         throw DataXException.asDataXException(KafkaReaderErrorCode.ILLEGAL_PARAM,
    212.                                 String.format("您提供的配置文件有误，datax不支持数据类型[%s]", columnTypes.get(i)));
    213.                 }
    214.             }
    215.         }
    216.     }
    217. }

    重点看 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

    1. {
    2.     "name": "kafkareader",
    3.     "class": "com.qsl.datax.plugin.reader.kafkareader.KafkaReader",
    4.     "description": "read data from kafka",
    5.     "developer": "qsl"
    6. }

    class 是 KafkaReader 的全限定类名

  3. 配置文件

    在 resources 下新增 plugin_job_template.json

    1. {
    2.     "name": "kafkareader",
    3.     "parameter": {
    4.         "bootstrapServers": "",
    5.         "topic": "test-kafka",
    6.         "groupId": "test1",
    7.         "writeType": "json",
    8.         "pollTimeoutMs": 2000,
    9.         "columnType": [
    10.             "LONG",
    11.             "STRING",
    12.             "STRING"
    13.         ],
    14.         "sasl": {
    15.             "securityProtocol": "SASL_PLAINTEXT",
    16.             "mechanism": "PLAIN",
    17.             "username": "",
    18.             "password": "2"
    19.         }
    20.     }
    21. }

    配置项说明：kafkareader.md

  4. 打包发布

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

至此，kafkareader 也完成了

总结

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

原文链接：https://www.cnblogs.com/youzhibing/p/18378097

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Apache Kafka热门文章

  • Kafka监控工具汇总
  • Kraft模式下Kafka脚本的使用
  • Kafka web console安装
  • kafka_2.11-2.0.0_安装部署
  • ELK日志方案--使用Filebeat收集日志并输出到Kafka
  • Kafka相关内容总结（存储和性能）
  • 一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
  • Kafka使用jmxtrans+influxdb+grafana监控JMX指标
  • Micronaut 微服务中使用 Kafka
  • 什么是Kafka？

    • Apache Kafka推荐文章

  • 异源数据同步 → DataX 为什么要支持 kafka？
  • 解锁强强组合: 使用 Kafka + ClickHouse 快速搭建流数据实时处理平台(DoubleCloud 博客)
  • Kafka 线上性能调优
  • Kafka原理剖析之「位点提交」
  • Kafka 社区KIP-382中文译文（MirrorMaker2/集群复制/高可用/灾难恢复）
  • kafka源码阅读之MacBook Pro M1搭建Kafka2.7版本源码运行环境
  • Kafka干货之「零拷贝」
  • 扫盲Kafka？看这一篇就够了！
  • 从一次Kafka宕机说起（JVM hang）
  • 从Kafka中学习高性能系统如何设计

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号