Send excel data to Kafka

Question

I would like some one to review my code and let me know the feedback.

Each Kafka Message is like as follows

[{guid=id.Value1, timestamp=1386394980000, booleanValue=null, longValue=null, floatValue=null, stringValue=value1, source=X, metadata=null}]

This is what I implemented:

ExcelReader

import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.xssf.usermodel.XSSFSheet;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;

import java.io.IOException;
import java.io.InputStream;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ExcelReader{

    public static List<KafkaMessage> readExcel(final InputStream inputStream) throws ParseException{

        List<KafkaMessage> messages = new ArrayList<>();
        XSSFWorkbook workbook;

        try{
            workbook = new XSSFWorkbook(inputStream);
            XSSFSheet sheet = workbook.getSheetAt(0);
            messages = readRows(sheet, messages);
            return messages;

        }catch(IOException e){
            e.printStackTrace();
        }

        return messages;

    }

    private static List<KafkaMessage> readRows(XSSFSheet sheet, List<KafkaMessage> rows) throws ParseException{

        for(int i = 3; i <= sheet.getLastRowNum() - 1; i++) {

            Row row = sheet.getRow(i);

            for(int j = 6; j <= row.getLastCellNum(); j++) {
                if(j == 6) {
                    KafkaMessage kafkaMessage = new KafkaMessage();
                    String str1 = row.getCell(5).getStringCellValue();
                    kafkaMessage.setTimestamp(convertToTimeStamp(str1));
                    String str2 = row.getCell(j).getStringCellValue();
                    kafkaMessage.setStringValue(str2);
                    kafkaMessage.setSource(sources().get(j));
                    kafkaMessage.setGuid(conditionParametersIds().get(j));
                    rows.add(kafkaMessage);
                }
                if(j == 7) {
                    KafkaMessage kafkaMessage = new KafkaMessage();
                    String str1 = row.getCell(5).getStringCellValue();
                    kafkaMessage.setTimestamp(convertToTimeStamp(str1));
                    double str3 = row.getCell(j).getNumericCellValue();
                    kafkaMessage.setLongValue((long) str3);
                    kafkaMessage.setSource(sources().get(j));
                    kafkaMessage.setGuid(conditionParametersIds().get(j));
                    rows.add(kafkaMessage);
                }
                if(j == 8) {
                    KafkaMessage kafkaMessage = new KafkaMessage();
                    String str1 = row.getCell(5).getStringCellValue();
                    kafkaMessage.setTimestamp(convertToTimeStamp(str1));
                    boolean str4 = row.getCell(j).getBooleanCellValue();
                    kafkaMessage.setBooleanValue(str4);
                    kafkaMessage.setSource(sources().get(j));
                    kafkaMessage.setGuid(conditionParametersIds().get(j));
                    rows.add(kafkaMessage);
                }
                if(j == 9) {
                    KafkaMessage kafkaMessage = new KafkaMessage();
                    String str1 = row.getCell(5).getStringCellValue();
                    kafkaMessage.setTimestamp(convertToTimeStamp(str1));
                    double str5 = row.getCell(j).getNumericCellValue();
                    kafkaMessage.setFloatValue((float) str5);
                    kafkaMessage.setSource(sources().get(j));
                    kafkaMessage.setGuid(conditionParametersIds().get(j));
                    rows.add(kafkaMessage);
                }
            }
        }
        return rows;
    }

    private static Long convertToTimeStamp(String dateTime) throws ParseException{
        return DateUtil.provideDateFormat().parse(dateTime).getTime();
    }

    private static Map<Integer, String> conditionParametersIds(){
        Map<Integer, String> map = new HashMap<>();
        MapHelper.repeatPut(map, new Integer[]{6, 7, 8, 9}, new String[]{"id.Value1", "id.Value2", "id.Value3", "id.Value4"});
        return map;

    }

    private static Map<Integer, String> sources(){
        Map<Integer, String> map = new HashMap<>();
        MapHelper.repeatPut(map, new Integer[]{6, 7}, "X");
        MapHelper.repeatPut(map, new Integer[]{8, 9}, "Y");
        return map;
    }

    public enum MapHelper{
        ; // Utility class for working with maps

        public static <K, V> void repeatPut(Map<? super K, ? super V> map, K[] keys, V value){
            for(K key : keys) {
                map.put(key, value);
            }
        }

        public static <K, V> void repeatPut(Map<? super K, ? super V> map, K[] keys, V[] values){
            for(int index = 0; index < Math.min(keys.length, values.length); index++) {
                map.put(keys[index], values[index]);
            }
        }
    }

}

SendKafkaMessage

import org.json.JSONArray;
import org.json.JSONObject;

import java.util.List;

public class SendKafkaMessage{
    public void sendToProducer(List<KafkaMessage> messages){
        Producer producer = new Producer();
        messages.forEach(msg -> producer.start(getMessage(msg)));
    }

    public String getMessage(KafkaMessage message){
        JSONObject obj= new JSONObject(message);
        JSONArray jsonArray = new JSONArray();
        return jsonArray.put(obj).toString();

    }
}

KafkaProducer

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;


public class Producer {
    private Properties properties = new Properties();
    String topicName = "topicname";

    public void start(String value){
        String bootstrapServer = "url";
        String keySerializer = StringSerializer.class.getName();
        String valueSerializer = StringSerializer.class.getName();
        String producerId = "simpleProducer";
        int retries = 2;

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, producerId);
        properties.put(ProducerConfig.RETRIES_CONFIG, retries);

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topicName, "1",value);
        kafkaProducer.send(producerRecord);
        kafkaProducer.close();
    }

}

Test Class

import com.fasterxml.jackson.core.JsonProcessingException;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.text.ParseException;



public class ExcelToJsonTest {
    public static void main(String[] args) throws FileNotFoundException, JsonProcessingException, ParseException{
      FileInputStream fis = new FileInputStream("FilePath");
        if(fis == null)
            System.out.println("fis null");
         SendKafkaMessage sendKafkaMessage = new SendKafkaMessage();
         sendKafkaMessage.sendToProducer(ExcelReader.readExcel(fis));
    }
}

````

dariosicily · Accepted Answer · 2019-10-27 19:12:29Z

I noticed in the code of your ExcelReader class the following methods that return two maps:

private static Map<Integer, String> conditionParametersIds(){
    Map<Integer, String> map = new HashMap<>();
    MapHelper.repeatPut(map, new Integer[]{6, 7, 8, 9}, new String[]{"id.Value1", "id.Value2", "id.Value3", "id.Value4"});
    return map;
}
private static Map<Integer, String> sources(){
    Map<Integer, String> map = new HashMap<>();
    MapHelper.repeatPut(map, new Integer[]{6, 7}, "X");
    MapHelper.repeatPut(map, new Integer[]{8, 9}, "Y");
    return map;
}

You are repeating calls to these two methods to retrieve values from maps, a possible improvement to avoid this issue is to define two static final maps in your class initialized by the same methods returning unmodifiable maps like the code below:

private static Map<Integer, String> conditionParametersIds(){
    Map<Integer, String> map = new HashMap<>();
    MapHelper.repeatPut(map, new Integer[]{6, 7, 8, 9}, new String[]{"id.Value1", "id.Value2", "id.Value3", "id.Value4"});
    return Collections.unmodifiableMap(map);
}

private static Map<Integer, String> sources(){
    Map<Integer, String> map = new HashMap<>();
    MapHelper.repeatPut(map, new Integer[]{6, 7}, "X");
    MapHelper.repeatPut(map, new Integer[]{8, 9}, "Y");
    return Collections.unmodifiableMap(map);
}

private static final Map<Integer, String> SOURCES = sources();
private static final Map<Integer, String> CONDITION_PARAMETER_IDS = conditionParametersIds();

I noticed you have a loop in the method readRows of ExcelReader class performing action on values in the range 6 ...9 and ignoring others out of this range:

for(int j = 6; j <= row.getLastCellNum(); j++) {
    if(j == 6) {
         //omitted
    }
    if(j == 7) {
        //omitted
    } //etc..
}

You can rewrite the loop calculating the minimal value between 9 and row.getLastCellNum(), using a switch inside the loop for values between 6 and 9 and peculiar actions like my code below:

int min = Math.min(row.getLastCellNum(), 9);

for(int j = 6; j <= min; j++) {
    KafkaMessage kafkaMessage = new KafkaMessage();

    //initialize the common fields to all j values
    String str = row.getCell(5).getStringCellValue();
    kafkaMessage.setTimestamp(convertToTimeStamp(str));
    kafkaMessage.setSource(SOURCES.get(j));
    kafkaMessage.setGuid(CONDITION_PARAMETER_IDS.get(j));

    switch(j) {
        case 6:
            str = row.getCell(j).getStringCellValue();
            kafkaMessage.setStringValue(str);
            break;
        case 7:
            double d = row.getCell(j).getNumericCellValue();
            kafkaMessage.setLongValue((long) d);
            break;
        case 8:
            boolean b = row.getCell(j).getBooleanCellValue();
            kafkaMessage.setBooleanValue(b);
            break;
        case 9:
            d = row.getCell(j).getNumericCellValue();
            kafkaMessage.setFloatValue((float) d);
            break;
    }

    rows.add(kafkaMessage);
}

I have seen that your readExcel method throws a ParseException but not IOException because you choose to catch the exception, for me the user of this method should be informed about the fact something was wrong with the file and so the exception should be thrown also.

Stack Exchange Network

Send excel data to Kafka

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Send excel data to Kafka

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions