• DataX二次开发——（6）kafkareader、kafkawriter的开发

    参考博客：

    Datax 二次开发插件详细过程_键盘上的艺术家w的博客-CSDN博客_datax kafkareader

    简书-DataX kafkawriter

     背景

    基于阿里开源DataX3.0版本，开发kafka的读写驱动，可以实现从mysql、postgresql抽取数据到kafka，从kafka 消费消息写入hdfs等功能。

    1、整体模块代码结构

    1、kafkareader

    2、kafkawriter 

    3、package.xml

    1.         <fileSet>
    2.             <directory>kafkareader/target/datax/</directory>
    3.             <includes>
    4.                 <include>**/*.*</include>
    5.             </includes>
    6.             <outputDirectory>datax</outputDirectory>
    7.         </fileSet>        
    8.         <fileSet>
    9.             <directory>kafkawriter/target/datax/</directory>
    10.             <includes>
    11.                 <include>**/*.*</include>
    12.             </includes>
    13.             <outputDirectory>datax</outputDirectory>
    14.         </fileSet>

    4、pom.xml

    1.         <module>kafkareader</module>
    2.         <module>kafkawriter</module>

    2、kafkareader模块代码

    package.xml

    1. <assembly
    2.         xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
    3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.         xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
    5.     <id></id>
    6.     <formats>
    7.         <format>dir</format>
    8.     </formats>
    9.     <includeBaseDirectory>false</includeBaseDirectory>
    10.     <fileSets>
    11.         <fileSet>
    12.             <directory>src/main/resources</directory>
    13.             <includes>
    14.                 <include>plugin.json</include>
    15.             </includes>
    16.             <outputDirectory>plugin/reader/kafkareader</outputDirectory>
    17.         </fileSet>
    18.         <fileSet>
    19.             <directory>target/</directory>
    20.             <includes>
    21.                 <include>kafkareader-0.0.1-SNAPSHOT.jar</include>
    22.             </includes>
    23.             <outputDirectory>plugin/reader/kafkareader</outputDirectory>
    24.         </fileSet>
    25.     </fileSets>
    26.  
    27.     <dependencySets>
    28.         <dependencySet>
    29.             <useProjectArtifact>false</useProjectArtifact>
    30.             <outputDirectory>plugin/reader/kafkareader/libs</outputDirectory>
    31.             <scope>runtime</scope>
    32.         </dependencySet>
    33.     </dependencySets>
    34. </assembly>

    DateUtil.java 

    1. package com.alibaba.datax.plugin.reader.kafkareader;
    2.  
    3. import java.text.SimpleDateFormat;
    4. import java.util.Date;
    5.  
    6. /**
    7.  * @Description: TODO
    8.  * @Author: chenweifeng
    9.  * @Date: 2022年09月05日 下午1:43
    10.  **/
    11. public class DateUtil {
    12.  
    13.     /**
    14.      * 时间转 yyyy-MM-dd HH:mm:ss 字符串
    15.      *
    16.      * @param date
    17.      * @return
    18.      */
    19.     public static String targetFormat(Date date) {
    20.         String dateString = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
    21.         return dateString;
    22.     }
    23.  
    24.     /**
    25.      * 时间转字符串
    26.      *
    27.      * @param date
    28.      * @param pattern
    29.      * @return
    30.      */
    31.     public static String targetFormat(Date date, String pattern) {
    32.         String dateString = new SimpleDateFormat(pattern).format(date);
    33.         return dateString;
    34.     }
    35. }

     JsonUtilJava.java

    1. package com.alibaba.datax.plugin.reader.kafkareader;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.alibaba.fastjson.TypeReference;
    5.  
    6. import java.util.HashMap;
    7.  
    8. /**
    9.  * @Description: TODO
    10.  * @Author: chenweifeng
    11.  * @Date: 2022年09月05日 下午1:47
    12.  **/
    13. public class JsonUtilJava {
    14.  
    15.     /**
    16.      * json转hashmap
    17.      *
    18.      * @param jsonData
    19.      * @return
    20.      */
    21.     public static HashMap<String, Object> parseJsonStrToMap(String jsonData) {
    22.         HashMap<String, Object> hashMap = JSONObject.parseObject(jsonData, new TypeReference<HashMap<String, Object>>() {
    23.         });
    24.         return hashMap;
    25.     }
    26. }

    KafkaReader.java

    1. package com.alibaba.datax.plugin.reader.kafkareader;
    2.  
    3. import com.alibaba.datax.common.element.Record;
    4. import com.alibaba.datax.common.element.StringColumn;
    5. import com.alibaba.datax.common.exception.DataXException;
    6. import com.alibaba.datax.common.plugin.RecordSender;
    7. import com.alibaba.datax.common.spi.Reader;
    8. import com.alibaba.datax.common.util.Configuration;
    9. import org.apache.kafka.clients.consumer.ConsumerConfig;
    10. import org.apache.kafka.clients.consumer.ConsumerRecord;
    11. import org.apache.kafka.clients.consumer.ConsumerRecords;
    12. import org.apache.kafka.clients.consumer.KafkaConsumer;
    13. import org.slf4j.Logger;
    14. import org.slf4j.LoggerFactory;
    15.  
    16. import java.io.FileNotFoundException;
    17. import java.io.FileOutputStream;
    18. import java.io.IOException;
    19. import java.time.Duration;
    20. import java.util.*;
    21. import java.util.regex.Matcher;
    22. import java.util.regex.Pattern;
    23.  
    24. public class KafkaReader extends Reader {
    25.  
    26.     public static class Job extends Reader.Job {
    27.         private static final Logger LOG = LoggerFactory.getLogger(Job.class);
    28.  
    29.         private Configuration originalConfig = null;
    30.  
    31.  
    32.         @Override
    33.         public void init() {
    34.             this.originalConfig = super.getPluginJobConf();
    35.             // warn: 忽略大小写
    36.  
    37.             String topic = this.originalConfig.getString(Key.TOPIC);
    38.             Integer partitions = this.originalConfig.getInt(Key.KAFKA_PARTITIONS);
    39.             String bootstrapServers = this.originalConfig.getString(Key.BOOTSTRAP_SERVERS);
    40.             String groupId = this.originalConfig.getString(Key.GROUP_ID);
    41.             Integer columnCount = this.originalConfig.getInt(Key.COLUMNCOUNT);
    42.             String split = this.originalConfig.getString(Key.SPLIT);
    43.             String filterContaintsStr = this.originalConfig.getString(Key.CONTAINTS_STR);
    44.             String filterContaintsFlag = this.originalConfig.getString(Key.CONTAINTS_STR_FLAG);
    45.             String conditionAllOrOne = this.originalConfig.getString(Key.CONDITION_ALL_OR_ONE);
    46.             String parsingRules = this.originalConfig.getString(Key.PARSING_RULES);
    47.             String writerOrder = this.originalConfig.getString(Key.WRITER_ORDER);
    48.             String kafkaReaderColumnKey = this.originalConfig.getString(Key.KAFKA_READER_COLUMN_KEY);
    49.  
    50.             LOG.info("topic:{},partitions:{},bootstrapServers:{},groupId:{},columnCount:{},split:{},parsingRules:{}",
    51.                     topic, partitions, bootstrapServers, groupId, columnCount, split, parsingRules);
    52.  
    53.             if (null == topic) {
    54.                 throw DataXException.asDataXException(KafkaReaderErrorCode.TOPIC_ERROR,
    55.                         "没有设置参数[topic].");
    56.             }
    57.             if (partitions == null) {
    58.                 throw DataXException.asDataXException(KafkaReaderErrorCode.PARTITION_ERROR,
    59.                         "没有设置参数[kafka.partitions].");
    60.             } else if (partitions < 1) {
    61.                 throw DataXException.asDataXException(KafkaReaderErrorCode.PARTITION_ERROR,
    62.                         "[kafka.partitions]不能小于1.");
    63.             }
    64.             if (null == bootstrapServers) {
    65.                 throw DataXException.asDataXException(KafkaReaderErrorCode.ADDRESS_ERROR,
    66.                         "没有设置参数[bootstrap.servers].");
    67.             }
    68.             if (null == groupId) {
    69.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    70.                         "没有设置参数[groupid].");
    71.             }
    72.             if (columnCount == null) {
    73.                 throw DataXException.asDataXException(KafkaReaderErrorCode.PARTITION_ERROR,
    74.                         "没有设置参数[columnCount].");
    75.             } else if (columnCount < 1) {
    76.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    77.                         "[columnCount]不能小于1.");
    78.             }
    79.             if (null == split) {
    80.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    81.                         "[split]不能为空.");
    82.             }
    83.             if (filterContaintsStr != null) {
    84.                 if (conditionAllOrOne == null || filterContaintsFlag == null) {
    85.                     throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    86.                             "设置了[filterContaintsStr],但是没有设置[conditionAllOrOne]或者[filterContaintsFlag]");
    87.                 }
    88.             }
    89.             if (parsingRules == null) {
    90.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    91.                         "没有设置[parsingRules]参数");
    92.             } else if (!parsingRules.equals("regex") && parsingRules.equals("json") && parsingRules.equals("split")) {
    93.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    94.                         "[parsingRules]参数设置错误，不是regex，json，split其中一个");
    95.             }
    96.             if (writerOrder == null) {
    97.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    98.                         "没有设置[writerOrder]参数");
    99.             }
    100.             if (kafkaReaderColumnKey == null) {
    101.                 throw DataXException.asDataXException(KafkaReaderErrorCode.KAFKA_READER_ERROR,
    102.                         "没有设置[kafkaReaderColumnKey]参数");
    103.             }
    104.         }
    105.  
    106.         @Override
    107.         public void preCheck() {
    108.             init();
    109.         }
    110.  
    111.         @Override
    112.         public List<Configuration> split(int adviceNumber) {
    113.             List<Configuration> configurations = new ArrayList<Configuration>();
    114.  
    115.             Integer partitions = this.originalConfig.getInt(Key.KAFKA_PARTITIONS);
    116.             for (int i = 0; i < partitions; i++) {
    117.                 configurations.add(this.originalConfig.clone());
    118.             }
    119.             return configurations;
    120.         }
    121.  
    122.         @Override
    123.         public void post() {
    124.         }
    125.  
    126.         @Override
    127.         public void destroy() {
    128.  
    129.         }
    130.  
    131.     }
    132.  
    133.     public static class Task extends Reader.Task {
    134.  
    135.         private static final Logger LOG = LoggerFactory.getLogger(Task.class);
    136.         //配置文件
    137.         private Configuration readerSliceConfig;
    138.         //kafka消息的分隔符
    139.         private String split;
    140.         //解析规则
    141.         private String parsingRules;
    142.         //是否停止拉去数据
    143.         private boolean flag;
    144.         //kafka address
    145.         private String bootstrapServers;
    146.         //kafka groupid
    147.         private String groupId;
    148.         //kafkatopic
    149.         private String kafkaTopic;
    150.         //kafka中的数据一共有多少个字段
    151.         private int count;
    152.         //是否需要data_from
    153.         //kafka ip 端口+ topic
    154.         //将包含/不包含该字符串的数据过滤掉
    155.         private String filterContaintsStr;
    156.         //是包含containtsStr 还是不包含
    157.         //1 表示包含 0 表示不包含
    158.         private int filterContaintsStrFlag;
    159.         //全部包含或不包含，包含其中一个或者不包含其中一个。
    160.         private int conditionAllOrOne;
    161.         //writer端要求的顺序。
    162.         private String writerOrder;
    163.         //kafkareader端的每个关键子的key
    164.         private String kafkaReaderColumnKey;
    165.         //异常文件路径
    166.         private String exceptionPath;
    167.         //
    168.         private Boolean enableAutoCommit;
    169.         //
    170.         private Integer maxPollRecords;
    171.  
    172.         @Override
    173.         public void init() {
    174.             flag = true;
    175.             this.readerSliceConfig = super.getPluginJobConf();
    176.             split = this.readerSliceConfig.getString(Key.SPLIT);
    177.             bootstrapServers = this.readerSliceConfig.getString(Key.BOOTSTRAP_SERVERS);
    178.             groupId = this.readerSliceConfig.getString(Key.GROUP_ID);
    179.             kafkaTopic = this.readerSliceConfig.getString(Key.TOPIC);
    180.             count = this.readerSliceConfig.getInt(Key.COLUMNCOUNT);
    181.             filterContaintsStr = this.readerSliceConfig.getString(Key.CONTAINTS_STR);
    182.             filterContaintsStrFlag = this.readerSliceConfig.getInt(Key.CONTAINTS_STR_FLAG);
    183.             conditionAllOrOne = this.readerSliceConfig.getInt(Key.CONTAINTS_STR_FLAG);
    184.             parsingRules = this.readerSliceConfig.getString(Key.PARSING_RULES);
    185.             writerOrder = this.readerSliceConfig.getString(Key.WRITER_ORDER);
    186.             kafkaReaderColumnKey = this.readerSliceConfig.getString(Key.KAFKA_READER_COLUMN_KEY);
    187.             exceptionPath = this.readerSliceConfig.getString(Key.EXECPTION_PATH);
    188.             enableAutoCommit = Objects.isNull(this.readerSliceConfig.getBool(Key.ENABLE_AUTO_COMMIT)) ? true : this.readerSliceConfig.getBool(Key.EXECPTION_PATH);
    189.             maxPollRecords = Objects.isNull(this.readerSliceConfig.getInt(Key.MAX_POLL_RECORDS)) ? 200 : this.readerSliceConfig.getInt(Key.MAX_POLL_RECORDS);
    190.             LOG.info(filterContaintsStr);
    191.         }
    192.  
    193.         @Override
    194.         public void startRead(RecordSender recordSender) {
    195.  
    196.             Properties props = new Properties();
    197.             props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    198.             props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId != null ? groupId : UUID.randomUUID().toString());
    199.             props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    200.             props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    201.             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
    202.             props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
    203.  
    204.  
    205.             KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    206.             consumer.subscribe(Collections.singletonList(kafkaTopic));
    207.  
    208.             Record oneRecord = null;
    209.             int commitSyncMaxNum = maxPollRecords;
    210.             int commitSyncNum = 0;
    211.             while (flag) {
    212.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    213.                 for (ConsumerRecord<String, String> record : records) {
    214.  
    215.                     String value = record.value();
    216.                     //定义过滤标志
    217.                     int ifNotContinue = filterMessage(value);
    218.                     //如果标志修改为1了那么就过滤掉这条数据。
    219.                     if (ifNotContinue == 1) {
    220.                         LOG.info("过滤数据: " + record.value());
    221.                         continue;
    222.                     }
    223.                     oneRecord = buildOneRecord(recordSender, value);
    224.                     //如果返回值不等于null表示不是异常消息。
    225.                     if (oneRecord != null) {
    226.                         LOG.info("oneRecord:{}", oneRecord.toString());
    227.                         recordSender.sendToWriter(oneRecord);
    228.                     }
    229.                 }
    230.                 recordSender.flush();
    231.  
    232.                 if (!enableAutoCommit) {
    233.                     commitSyncNum++;
    234.                     if (commitSyncNum >= commitSyncMaxNum) {
    235.                         consumer.commitSync();
    236.                         commitSyncNum = 0;
    237.                     }
    238.                 }
    239.  
    240.                 //判断当前事件是不是0点,0点的话进程他退出
    241.                 Date date = new Date();
    242.                 if (DateUtil.targetFormat(date).split(" ")[1].substring(0, 2).equals("00")) {
    243.                     destroy();
    244.                 }
    245.  
    246.             }
    247.         }
    248.  
    249.         private int filterMessage(String value) {
    250.             //如果要过滤的条件配置了
    251.             int ifNotContinue = 0;
    252.  
    253.             if (filterContaintsStr != null) {
    254.                 String[] filterStrs = filterContaintsStr.split(",");
    255.                 //所有
    256.                 if (conditionAllOrOne == 1) {
    257.                     //过滤掉包含filterContaintsStr的所有项的值。
    258.                     if (filterContaintsStrFlag == 1) {
    259.                         int i = 0;
    260.                         for (; i < filterStrs.length; i++) {
    261.                             if (!value.contains(filterStrs[i])) break;
    262.                         }
    263.                         if (i >= filterStrs.length) ifNotContinue = 1;
    264.                     } else {
    265.                         //留下掉包含filterContaintsStr的所有项的值
    266.                         int i = 0;
    267.                         for (; i < filterStrs.length; i++) {
    268.                             if (!value.contains(filterStrs[i])) break;
    269.                         }
    270.                         if (i < filterStrs.length) ifNotContinue = 1;
    271.                     }
    272.  
    273.                 } else {
    274.                     //过滤掉包含其中一项的值
    275.                     if (filterContaintsStrFlag == 1) {
    276.                         int i = 0;
    277.                         for (; i < filterStrs.length; i++) {
    278.                             if (value.contains(filterStrs[i])) break;
    279.                         }
    280.                         if (i < filterStrs.length) ifNotContinue = 1;
    281.                     }
    282.                     //留下包含其中一下的值
    283.                     else {
    284.                         int i = 0;
    285.                         for (; i < filterStrs.length; i++) {
    286.                             if (value.contains(filterStrs[i])) break;
    287.                         }
    288.                         if (i >= filterStrs.length) ifNotContinue = 1;
    289.                     }
    290.                 }
    291.             }
    292.             return ifNotContinue;
    293.  
    294.         }
    295.  
    296.         private Record buildOneRecord(RecordSender recordSender, String value) {
    297.             Record record = null;
    298.             if (parsingRules.equals("regex")) {
    299.                 record = parseRegex(value, recordSender);
    300.             } else if (parsingRules.equals("json")) {
    301.                 record = parseJson(value, recordSender);
    302.             } else if (parsingRules.equals("split")) {
    303.                 record = parseSplit(value, recordSender);
    304.             }
    305.             LOG.info("record:{}", record.toString());
    306.             return record;
    307.         }
    308.  
    309.         private Record parseSplit(String value, RecordSender recordSender) {
    310.             Record record = recordSender.createRecord();
    311.             String[] splits = value.split(this.split);
    312.             if (splits.length != count) {
    313.                 writerErrorPath(value);
    314.                 return null;
    315.             }
    316.             return parseOrders(Arrays.asList(splits), record);
    317.         }
    318.  
    319.         private Record parseJson(String value, RecordSender recordSender) {
    320.             LOG.info("parseJson value :{}", value);
    321.             Record record = recordSender.createRecord();
    322.             HashMap<String, Object> map = JsonUtilJava.parseJsonStrToMap(value);
    323.             LOG.info("map :{}", map);
    324.             List<Map<String, Object>> mapData = (List>) map.get("data");
    325.             LOG.info("mapData :{}", mapData);
    326.  
    327.             LOG.info("parseJson kafkaReaderColumnKey :{}", kafkaReaderColumnKey);
    328.             String[] columns = kafkaReaderColumnKey.split(",");
    329.  
    330.             if (mapData.size() != columns.length) {
    331.                 throw new RuntimeException("kafka字段数和columns的字段数不一致，无法映射数据");
    332.             }
    333.  
    334.             ArrayList<String> datas = new ArrayList<>();
    335.             for (int i = 0; i < columns.length; i++) {
    336.                 datas.add(String.valueOf(mapData.get(i).get("rawData")));
    337.             }
    338.  
    339. //            for (String column : columns) {
    340. //                datas.add(map.get(column).toString());
    341. //            }
    342.             if (datas.size() != count) {
    343.                 writerErrorPath(value);
    344.                 return null;
    345.             }
    346.             LOG.info("datas:{}", datas);
    347.             return parseOrders(datas, record);
    348.         }
    349.  
    350.         private Record parseRegex(String value, RecordSender recordSender) {
    351.             Record record = recordSender.createRecord();
    352.             ArrayList<String> datas = new ArrayList<String>();
    353.             Pattern r = Pattern.compile(split);
    354.             Matcher m = r.matcher(value);
    355.             if (m.find()) {
    356.                 if (m.groupCount() != count) {
    357.                     writerErrorPath(value);
    358.                 }
    359.                 for (int i = 1; i <= count; i++) {
    360.                     //  record.addColumn(new StringColumn(m.group(i)));
    361.                     datas.add(m.group(i));
    362.                     return record;
    363.                 }
    364.             } else {
    365.                 writerErrorPath(value);
    366.             }
    367.  
    368.             return parseOrders(datas, record);
    369.         }
    370.  
    371.         private void writerErrorPath(String value) {
    372.             if (exceptionPath == null) return;
    373.             FileOutputStream fileOutputStream = null;
    374.             try {
    375.                 fileOutputStream = getFileOutputStream();
    376.                 fileOutputStream.write((value + "\n").getBytes());
    377.                 fileOutputStream.close();
    378.             } catch (FileNotFoundException e) {
    379.                 e.printStackTrace();
    380.             } catch (IOException e) {
    381.                 e.printStackTrace();
    382.             }
    383.         }
    384.  
    385.         private FileOutputStream getFileOutputStream() throws FileNotFoundException {
    386.             return new FileOutputStream(exceptionPath + "/" + kafkaTopic + "errordata" + DateUtil.targetFormat(new Date(), "yyyyMMdd"), true);
    387.         }
    388.  
    389.         private Record parseOrders(List<String> datas, Record record) {
    390.             //writerOrder
    391.             String[] orders = writerOrder.split(",");
    392.             LOG.info("writerOrder:{}", writerOrder);
    393.             for (String order : orders) {
    394.                 if (order.equals("data_from")) {
    395.                     record.addColumn(new StringColumn(bootstrapServers + "|" + kafkaTopic));
    396.                 } else if (order.equals("uuid")) {
    397.                     record.addColumn(new StringColumn(UUID.randomUUID().toString()));
    398.                 } else if (order.equals("null")) {
    399.                     record.addColumn(new StringColumn("null"));
    400.                 } else if (order.equals("datax_time")) {
    401.                     record.addColumn(new StringColumn(DateUtil.targetFormat(new Date())));
    402.                 } else if (isNumeric(order)) {
    403.                     record.addColumn(new StringColumn(datas.get(new Integer(order) - 1)));
    404.                 }
    405.             }
    406.             return record;
    407.         }
    408.  
    409.         public static boolean isNumeric(String str) {
    410.             for (int i = 0; i < str.length(); i++) {
    411.                 if (!Character.isDigit(str.charAt(i))) {
    412.                     return false;
    413.                 }
    414.             }
    415.             return true;
    416.         }
    417.  
    418.         @Override
    419.         public void post() {
    420.         }
    421.  
    422.         @Override
    423.         public void destroy() {
    424.             flag = false;
    425.         }
    426.  
    427.  
    428.     }
    429. }

     KafkaReaderErrorCode.java

    1. package com.alibaba.datax.plugin.reader.kafkareader;
    2.  
    3. import com.alibaba.datax.common.spi.ErrorCode;
    4.  
    5. /**
    6.  * @Description: TODO
    7.  * @Author: chenweifeng
    8.  * @Date: 2022年09月05日 上午11:14
    9.  **/
    10. public enum KafkaReaderErrorCode implements ErrorCode {
    11.     TOPIC_ERROR("KafkaReader-00", "您没有设置topic参数"),
    12.     PARTITION_ERROR("KafkaReader-01", "您没有设置kafkaPartitions参数"),
    13.     ADDRESS_ERROR("KafkaReader-02", "您没有设置bootstrapServers参数"),
    14.     KAFKA_READER_ERROR("KafkaReader-03", "参数错误"),
    15.     ;
    16.  
    17.  
    18.     private final String code;
    19.     private final String description;
    20.  
    21.     private KafkaReaderErrorCode(String code, String description) {
    22.         this.code = code;
    23.         this.description = description;
    24.     }
    25.  
    26.     @Override
    27.     public String getCode() {
    28.         return this.code;
    29.     }
    30.  
    31.     @Override
    32.     public String getDescription() {
    33.         return this.description;
    34.     }
    35.  
    36.     @Override
    37.     public String toString() {
    38.         return String.format("Code:[%s], Description:[%s].", this.code,
    39.                 this.description);
    40.     }
    41. }

    Key.java

    1. package com.alibaba.datax.plugin.reader.kafkareader;
    2.  
    3. /**
    4.  * @Description: TODO
    5.  * @Author: chenweifeng
    6.  * @Date: 2022年09月05日 上午11:08
    7.  **/
    8. public class Key {
    9.  
    10.     public final static String TOPIC = "topic";// 主题
    11.     public final static String KAFKA_PARTITIONS = "kafkaPartitions";// 分区数
    12.     public final static String BOOTSTRAP_SERVERS = "bootstrapServers";//
    13.     public final static String GROUP_ID = "groupId";// 消费组
    14.     public final static String COLUMNCOUNT = "columnCount";// 字段数
    15.     public final static String SPLIT = "split";// 分割符
    16.     public final static String CONTAINTS_STR = "filterContaints";// 过滤的字符串 英文逗号隔开
    17.     public final static String CONTAINTS_STR_FLAG = "filterContaintsFlag";// 1 表示包含 0 表示不包含
    18.     public final static String CONDITION_ALL_OR_ONE = "conditionAllOrOne";// 1 全部包含或不包含，0 包含其中一个或者不包含其中一个
    19.     public final static String PARSING_RULES = "parsingRules";//
    20.     public final static String WRITER_ORDER = "writerOrder";//
    21.     public final static String KAFKA_READER_COLUMN_KEY = "kafkaReaderColumnKey";//
    22.     public final static String EXECPTION_PATH = "execptionPath";//
    23.     public final static String ENABLE_AUTO_COMMIT = "enableAutoCommit";// 是否自动提交
    24.     public final static String MAX_POLL_RECORDS = "maxPollRecords";// 最大Poll数
    25.  
    26. }

    plugin.json 

    1. {
    2.     "name": "kafkareader",
    3.     "class": "com.alibaba.datax.plugin.reader.kafkareader.KafkaReader",
    4.     "description": "kafka reader 插件",
    5.     "developer": "chenweifeng"
    6. }

    plugin_job_template.json 

    1. {
    2.   "name": "kafkareader",
    3.   "parameter": {
    4.     "topic": "Event",
    5.     "bootstrapServers": "192.168.7.128:9092",
    6.     "kafkaPartitions": "1",
    7.     "columnCount": 11,
    8.     "groupId": "ast",
    9.     "filterContaints": "5^1,6^5",
    10.     "filterContaintsFlag": 1,
    11.     "conditionAllOrOne": 0,
    12.     "parsingRules": "regex",
    13.     "writerOrder": "uuid,1,3,6,4,8,9,10,11,5,7,2,null,datax_time,data_from",
    14.     "kafkaReaderColumnKey": "a",
    15.     "execptionPath": "/opt/module/datax/log/errorlog"
    16.   }
    17. }

    3、kafkawriter模块代码

    package.xml

    1. <assembly
    2.         xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
    3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.         xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
    5.     <id></id>
    6.     <formats>
    7.         <format>dir</format>
    8.     </formats>
    9.     <includeBaseDirectory>false</includeBaseDirectory>
    10.     <fileSets>
    11.         <fileSet>
    12.             <directory>src/main/resources</directory>
    13.             <includes>
    14.                 <include>plugin.json</include>
    15.             </includes>
    16.             <outputDirectory>plugin/writer/kafkawriter</outputDirectory>
    17.         </fileSet>
    18.         <fileSet>
    19.             <directory>target/</directory>
    20.             <includes>
    21.                 <include>kafkawriter-0.0.1-SNAPSHOT.jar</include>
    22.             </includes>
    23.             <outputDirectory>plugin/writer/kafkawriter</outputDirectory>
    24.         </fileSet>
    25.     </fileSets>
    26.  
    27.     <dependencySets>
    28.         <dependencySet>
    29.             <useProjectArtifact>false</useProjectArtifact>
    30.             <outputDirectory>plugin/writer/kafkawriter/libs</outputDirectory>
    31.             <scope>runtime</scope>
    32.         </dependencySet>
    33.     </dependencySets>
    34. </assembly>

    KafkaWriter.java

    1. package com.alibaba.datax.plugin.writer.kafkawriter;
    2.  
    3. import com.alibaba.datax.common.element.Column;
    4. import com.alibaba.datax.common.element.Record;
    5. import com.alibaba.datax.common.exception.DataXException;
    6. import com.alibaba.datax.common.plugin.RecordReceiver;
    7. import com.alibaba.datax.common.spi.Writer;
    8. import com.alibaba.datax.common.util.Configuration;
    9. import org.apache.kafka.clients.admin.AdminClient;
    10. import org.apache.kafka.clients.admin.ListTopicsResult;
    11. import org.apache.kafka.clients.admin.NewTopic;
    12. import org.apache.kafka.clients.producer.KafkaProducer;
    13. import org.apache.kafka.clients.producer.Producer;
    14. import org.apache.kafka.clients.producer.ProducerRecord;
    15. import org.slf4j.Logger;
    16. import org.slf4j.LoggerFactory;
    17.  
    18. import java.util.ArrayList;
    19. import java.util.List;
    20. import java.util.Properties;
    21.  
    22. /**
    23.  * @ClassName: KafkaWriter
    24.  * @Author: majun
    25.  * @CreateDate: 2019/2/20 11:06
    26.  * @Version: 1.0
    27.  * @Description: datax kafkawriter
    28.  */
    30. public class KafkaWriter extends Writer {
    31.  
    32.     public static class Job extends Writer.Job {
    33.  
    34.         private static final Logger logger = LoggerFactory.getLogger(Job.class);
    35.         private Configuration conf = null;
    36.  
    37.         @Override
    38.         public List<Configuration> split(int mandatoryNumber) {
    39.             List<Configuration> configurations = new ArrayList<Configuration>(mandatoryNumber);
    40.             for (int i = 0; i < mandatoryNumber; i++) {
    41.                 configurations.add(conf);
    42.             }
    43.             return configurations;
    44.         }
    45.  
    46.         private void validateParameter() {
    47.             this.conf.getNecessaryValue(Key.BOOTSTRAP_SERVERS, KafkaWriterErrorCode.REQUIRED_VALUE);
    48.             this.conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
    49.         }
    50.  
    51.         @Override
    52.         public void init() {
    53.             this.conf = super.getPluginJobConf();
    54.             logger.info("kafka writer params:{}", conf.toJSON());
    55.             this.validateParameter();
    56.         }
    57.  
    58.  
    59.         @Override
    60.         public void destroy() {
    61.  
    62.         }
    63.     }
    64.  
    65.     public static class Task extends Writer.Task {
    66.         private static final Logger logger = LoggerFactory.getLogger(Task.class);
    67.         private static final String NEWLINE_FLAG = System.getProperty("line.separator", "\n");
    68.  
    69.         private Producer<String, String> producer;
    70.         private String fieldDelimiter;
    71.         private Configuration conf;
    72.         private Properties props;
    73.  
    74.         @Override
    75.         public void init() {
    76.             this.conf = super.getPluginJobConf();
    77.             fieldDelimiter = conf.getUnnecessaryValue(Key.FIELD_DELIMITER, "\t", null);
    78.  
    79.             props = new Properties();
    80.             props.put("bootstrap.servers", conf.getString(Key.BOOTSTRAP_SERVERS));
    81.             props.put("acks", conf.getUnnecessaryValue(Key.ACK, "0", null));//这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证。
    82.             props.put("retries", conf.getUnnecessaryValue(Key.RETRIES, "0", null));
    83.             // Controls how much bytes sender would wait to batch up before publishing to Kafka.
    84.             //控制发送者在发布到kafka之前等待批处理的字节数。
    85.             //控制发送者在发布到kafka之前等待批处理的字节数。 满足batch.size和ling.ms之一，producer便开始发送消息
    86.             //默认16384   16kb
    87.             props.put("batch.size", conf.getUnnecessaryValue(Key.BATCH_SIZE, "16384", null));
    88.             props.put("linger.ms", 1);
    89.             props.put("key.serializer", conf.getUnnecessaryValue(Key.KEYSERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
    90.             props.put("value.serializer", conf.getUnnecessaryValue(Key.VALUESERIALIZER, "org.apache.kafka.common.serialization.StringSerializer", null));
    91.             producer = new KafkaProducer<String, String>(props);
    92.         }
    93.  
    94.         @Override
    95.         public void prepare() {
    96.             if (Boolean.valueOf(conf.getUnnecessaryValue(Key.NO_TOPIC_CREATE, "false", null))) {
    97.  
    98.                 ListTopicsResult topicsResult = AdminClient.create(props).listTopics();
    99.                 String topic = conf.getNecessaryValue(Key.TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE);
    100.  
    101.                 try {
    102.                     if (!topicsResult.names().get().contains(topic)) {
    103.                         new NewTopic(
    104.                                 topic,
    105.                                 Integer.valueOf(conf.getUnnecessaryValue(Key.TOPIC_NUM_PARTITION, "1", null)),
    106.                                 Short.valueOf(conf.getUnnecessaryValue(Key.TOPIC_REPLICATION_FACTOR, "1", null))
    107.                         );
    108.                         List<NewTopic> newTopics = new ArrayList<NewTopic>();
    109.                         AdminClient.create(props).createTopics(newTopics);
    110.                     }
    111.                 } catch (Exception e) {
    112.                     throw new DataXException(KafkaWriterErrorCode.CREATE_TOPIC, KafkaWriterErrorCode.REQUIRED_VALUE.getDescription());
    113.                 }
    114.             }
    115.         }
    116.  
    117.         @Override
    118.         public void startWrite(RecordReceiver lineReceiver) {
    119.             logger.info("start to writer kafka");
    120.             Record record = null;
    121.             while ((record = lineReceiver.getFromReader()) != null) {//说明还在读取数据,或者读取的数据没处理完
    122.                 //获取一行数据，按照指定分隔符 拼成字符串 发送出去
    123.                 if (conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null).toLowerCase()
    124.                         .equals(WriteType.TEXT.name().toLowerCase())) {
    125.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
    126.                             recordToString(record),
    127.                             recordToString(record))
    128.                     );
    129.                 } else if (conf.getUnnecessaryValue(Key.WRITE_TYPE, WriteType.TEXT.name(), null).toLowerCase()
    130.                         .equals(WriteType.JSON.name().toLowerCase())) {
    131.                     producer.send(new ProducerRecord<String, String>(this.conf.getString(Key.TOPIC),
    132.                             recordToString(record),
    133.                             record.toString())
    134.                     );
    135.                 }
    136.                 logger.info("complete write " + record.toString());
    137.                 producer.flush();
    138.             }
    139.         }
    140.  
    141.         @Override
    142.         public void destroy() {
    143.             if (producer != null) {
    144.                 producer.close();
    145.             }
    146.         }
    147.  
    148.         /**
    149.          * 数据格式化
    150.          *
    151.          * @param record
    152.          * @return
    153.          */
    154.         private String recordToString(Record record) {
    155.             int recordLength = record.getColumnNumber();
    156.             if (0 == recordLength) {
    157.                 return NEWLINE_FLAG;
    158.             }
    159.             Column column;
    160.             StringBuilder sb = new StringBuilder();
    161.             for (int i = 0; i < recordLength; i++) {
    162.                 column = record.getColumn(i);
    163.                 sb.append(column.asString()).append(fieldDelimiter);
    164.             }
    165.  
    166.             sb.setLength(sb.length() - 1);
    167.             sb.append(NEWLINE_FLAG);
    168.  
    169.             logger.info("recordToString:{}",sb.toString());
    170.             return sb.toString();
    171.         }
    172.     }
    173. }

    KafkaWriterErrorCode.java

    1. package com.alibaba.datax.plugin.writer.kafkawriter;
    2.  
    3. import com.alibaba.datax.common.spi.ErrorCode;
    4.  
    5. public enum KafkaWriterErrorCode implements ErrorCode {
    6.  
    7.     REQUIRED_VALUE("KafkaWriter-00", "您缺失了必须填写的参数值."),
    8.     CREATE_TOPIC("KafkaWriter-01", "写入数据前检查topic或是创建topic失败.");
    9.  
    10.     private final String code;
    11.     private final String description;
    12.  
    13.     private KafkaWriterErrorCode(String code, String description) {
    14.         this.code = code;
    15.         this.description = description;
    16.     }
    17.  
    18.     @Override
    19.     public String getCode() {
    20.         return this.code;
    21.     }
    22.  
    23.     @Override
    24.     public String getDescription() {
    25.         return this.description;
    26.     }
    27.  
    28.     @Override
    29.     public String toString() {
    30.         return String.format("Code:[%s], Description:[%s].", this.code,
    31.                 this.description);
    32.     }
    33.  
    34. }

    Key.java

    1. package com.alibaba.datax.plugin.writer.kafkawriter;
    2.  
    3. /**
    4.  * @ClassName: Key
    5.  * @Author: majun
    6.  * @CreateDate: 2019/2/20 11:17
    7.  * @Version: 1.0
    8.  * @Description: TODO
    9.  */
    10.  
    11. public class Key {
    12. //
    13. //    bootstrapServers": "",
    14. //            "topic": "",
    15. //            "ack": "all",
    16. //            "batchSize": 1000,
    17. //            "retries": 0,
    18. //            "keySerializer":"org.apache.kafka.common.serialization.StringSerializer",
    19. //            "valueSerializer": "org.apache.kafka.common.serialization.StringSerializer",
    20. //            "fieldFelimiter": ","
    21.  
    22.     public static final String BOOTSTRAP_SERVERS = "bootstrapServers";
    23.  
    24.     // must have
    25.     public static final String TOPIC = "topic";
    26.  
    27.     public static final String ACK = "ack";
    28.  
    29.     public static final String BATCH_SIZE = "batchSize";
    30.  
    31.     public static final String RETRIES = "retries";
    32.  
    33.     public static final String KEYSERIALIZER = "keySerializer";
    34.  
    35.     public static final String VALUESERIALIZER = "valueSerializer";
    36.  
    37.     // not must , not default
    38.     public static final String FIELD_DELIMITER = "fieldDelimiter";
    39.  
    40.     public static final String NO_TOPIC_CREATE = "noTopicCreate";
    41.  
    42.     public static final String TOPIC_NUM_PARTITION = "topicNumPartition";
    43.  
    44.     public static final String TOPIC_REPLICATION_FACTOR = "topicReplicationFactor";
    45.  
    46.     public static final String WRITE_TYPE = "writeType";
    47. }

    WriterType.java

    1. package com.alibaba.datax.plugin.writer.kafkawriter;
    2.  
    3. public enum WriteType {
    4.     JSON("json"),
    5.     TEXT("text");
    6.  
    7.     private String name;
    8.  
    9.     WriteType(String name) {
    10.         this.name = name;
    11.     }
    12. }

    plugin.json 

    1. {
    2.   "name": "kafkawriter",
    3.   "class": "com.alibaba.datax.plugin.writer.kafkawriter.KafkaWriter",
    4.   "description": "kafka writer 插件",
    5.   "developer": "chenweifeng"
    6. }

    plugin_job_template.json  

    1. {
    2.   "name": "kafkawriter",
    3.   "parameter": {
    4.     "bootstrapServers": "10.1.20.150:9092",
    5.     "topic": "test-topic",
    6.     "ack": "all",
    7.     "batchSize": 1000,
    8.     "retries": 0,
    9.     "keySerializer": "org.apache.kafka.common.serialization.StringSerializer",
    10.     "valueSerializer": "org.apache.kafka.common.serialization.StringSerializer",
    11.     "fieldFelimiter": ",",
    12.     "writeType": "json",
    13.     "topicNumPartition": 1,
    14.     "topicReplicationFactor": 1
    15.   }
    16. }

    4、由kafka写到kafka的例子

    1. {
    2.     "job": {
    3.         "setting": {
    4.             "speed": {
    5.                  "channel": 12
    6.             },
    7.             "errorLimit": {
    8.                 "record": 0,
    9.                 "percentage": 0.02
    10.             }
    11.         },
    12.         "content": [
    13.             {
    14.                 "reader": {
    15.                       "name": "kafkareader",
    16.                       "parameter": {
    17.                         "topic": "test_datax_kafka_read",
    18.                         "bootstrapServers": "10.254.21.6:59292,10.254.21.1:59292,10.254.21.2:59292",
    19.                         "kafkaPartitions": 12,
    20.                         "columnCount": 34,
    21.                         "groupId": "datax_kafka_kafka",
    22.                         "filterContaints": "5^1,6^5",
    23.                         "filterContaintsFlag": 1,
    24.                         "conditionAllOrOne": 0,
    25.                         "parsingRules": "json",
    26.                         "writerOrder": "1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34",
    27.                         "kafkaReaderColumnKey": "id,c1,c2,c3,c4,c5,c6,c7,c8,c9,c10,c11,c12,c13,c14,c15,c16,c17,c18,c19,c20,c21,c22,c23,c24,c25,c26,c27,c28,c29,c30,c31,c32,crt_time",
    28.                         "execptionPath": "/Users/chenweifeng/datax/log/errorlog",
    29.                         "split":"\t"
    30.                       }
    31.                 },
    32.                 "writer": {
    33.                     "name": "kafkawriter",
    34.                     "parameter": {
    35.                         "print": true,
    36.                         "topic": "test_datax_kafka_write",
    37.                         "bootstrapServers": "10.254.21.6:59292,10.254.21.1:59292,10.254.21.2:59292",
    38.                         "fieldDelimiter": "\t",
    39.                         "batchSize": 20,
    40.                         "writeType": "json",
    41.                         "notTopicCreate": false,
    42.                         "topicNumPartition": 12,
    43.                         "topicReplicationFactor": 3
    44.               }
    45.           }
    46.             }
    47.         ]
    48.     }
    49. }
    50.

    注：目前kafkareader和kafkawriter处于可行状态，但是我本人做了性能测试，目前存在性能问题，做了一些参数的调整，仍然存在，以后会尝试进行优化，如有朋友已经对其进行优化的，可以在评论留下你建议，谢谢。

    性能问题的分析结果：DataX本身的架构说分批次进行读取，然后将数据经过Channel队列进行传输，和Kafka本身流式的数据是不兼容的，所以个人的建设是不太建议在DataX做kafka的数据同步，这个如果要解决这个问题就要破坏本身DataX的批式同步架构，所以后面本人放弃了DataX的Kafka数据源支持，改成在Flink实现Kafka的同步。

  • 相关阅读:
    Windows子系统WSL2 （ubuntu安装 docker、nvidia-docker）
    Scrum 敏捷管理流程图及敏捷管理工具
    GPT的历史
    Docker常用命令Top20
    vue中使用MINIO将文件上传到指定的bucket库中（vue2和vue3）
    【无标题】
    YUV和RGB的相互转换实验
    springboot+特色农产品电商平台 毕业设计-附源码211515
    【数据加密、解密】前后端数据传输的过程中，如何进行数据加密传输，保证数据的传输安全，防止被他人窃取
    基于篇章结构的英文作文自动评分方法（学习笔记）
  • 原文地址：https://blog.csdn.net/Carson073/article/details/126728411