將CSV的數據發送到kafka

為什麼將CSV的數據發到kafka

• flink做流式計算時，選用kafka消息作為數據源是常用手段，因此在學習和開發flink過程中，也會將數據集文件中的記錄發送到kafka，來模擬不間斷數據；
• 整個流程如下：

• 您可能會覺得這樣做多此一舉：flink直接讀取CSV不就行了嗎？這樣做的原因如下：
• 首先，這是學習和開發時的做法，數據集是CSV文件，而生產環境的實時數據卻是kafka數據源；
• 其次，Java應用中可以加入一些特殊邏輯，例如數據處理，彙總統計（用來和flink結果對比驗證）；
• 另外，如果兩條記錄實際的間隔時間如果是1分鐘，那麼Java應用在發送消息時也可以間隔一分鐘再發送，這個邏輯在flink社區的demo中有具體的實現，此demo也是將數據集發送到kafka，再由flink消費kafka，地址是：https://github.com/ververica/sql-training

如何將CSV的數據發送到kafka

前面的圖可以看出，讀取CSV再發送消息到kafka的操作是Java應用所為，因此今天的主要工作就是開發這個Java應用，並驗證；

版本信息

1. JDK：1.8.0_181
2. 開發工具：IntelliJ IDEA 2019.2.1 (Ultimate Edition)
3. 開發環境：Win10
4. Zookeeper：3.4.13
5. Kafka：2.4.0（scala：2.12）

關於數據集

• 本次實戰用到的數據集是CSV文件，裡面是一百零四萬條淘寶用戶行為數據，該數據來源是阿里雲天池公開數據集，我對此數據做了少量調整；
• 此CSV文件可以在CSDN下載，地址：

https://download.csdn.net/download/boling_cavalry/12381698

• 也可以在我的Github下載，地址：

https://raw.githubusercontent.com/zq2599/blog_demos/master/files/UserBehavior.7z

• 該CSV文件的內容，一共有六列，每列的含義如下表：

Java應用簡介

編碼前，先把具體內容列出來，然後再挨個實現：

1. 從CSV讀取記錄的工具類：UserBehaviorCsvFileReader
2. 每條記錄對應的Bean類：UserBehavior
3. Java對象序列化成JSON的序列化類：JsonSerializer
4. 向kafka發送消息的工具類：KafkaProducer
5. 應用類，程序入口：SendMessageApplication

上述五個類即可完成Java應用的工作，接下來開始編碼吧；

直接下載源碼

• 如果您不想編寫代碼，可以從我的Github下載完整源碼，地址是：https://github.com/zq2599/blog_demos
• blog_demos這個倉庫中有很多目錄，本次實戰源碼在flinksql目錄下，如下圖紅框所示：

編碼

• 創建maven工程，pom.xml如下，比較重要的jackson和javacsv的依賴：

<code>

    4.0.0

    com.bolingcavalry
    flinksql
    1.0-SNAPSHOT

    
        UTF-8
        1.10.0
        2.2.0
        1.8
        2.11
        ${java.version}
        ${java.version}
    

    
        
            org.apache.kafka
            kafka-clients
            ${kafka.version}
        

        
            com.fasterxml.jackson.core
            jackson-databind
            2.9.10.1
        

         
        
            org.slf4j
            slf4j-log4j12
            1.7.7
            runtime
        
        
            log4j
            log4j
            1.2.17
            runtime
        
        
            net.sourceforge.javacsv
            javacsv
            2.0
        

    

    
        
             
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.1
                
                    ${java.version}
                    ${java.version}
                
            

             
            
                org.apache.maven.plugins
                maven-shade-plugin
                3.0.0
                
                     
                    
                        package
                        
                            shade
                        
                        
                            
                                
                                
                            
                            
                                
                                     
                                    *:*
                                    
                                        META-INF/*.SF
                                        META-INF/*.DSA
                                        META-INF/*.RSA
                                    
                                
                            
                        
                    
                
            
        
    
/<code>

• 從CSV讀取記錄的工具類：UserBehaviorCsvFileReader，後面在主程序中會用到java8的Steam API來處理集合，所以UserBehaviorCsvFileReader實現了Supplier接口：

<code>public class UserBehaviorCsvFileReader implements Supplier {

    private final String filePath;
    private CsvReader csvReader;

    public UserBehaviorCsvFileReader(String filePath) throws IOException {

        this.filePath = filePath;
        try {
            csvReader = new CsvReader(filePath);
            csvReader.readHeaders();
        } catch (IOException e) {
            throw new IOException("Error reading TaxiRecords from file: " + filePath, e);
        }
    }

    @Override
    public UserBehavior get() {
        UserBehavior userBehavior = null;
        try{
            if(csvReader.readRecord()) {
                csvReader.getRawRecord();
                userBehavior = new UserBehavior(
                        Long.valueOf(csvReader.get(0)),
                        Long.valueOf(csvReader.get(1)),
                        Long.valueOf(csvReader.get(2)),
                        csvReader.get(3),
                        new Date(Long.valueOf(csvReader.get(4))*1000L));
            }
        } catch (IOException e) {
            throw new NoSuchElementException("IOException from " + filePath);
        }

        if (null==userBehavior) {
            throw new NoSuchElementException("All records read from " + filePath);
        }

        return userBehavior;
    }
}/<code>

• 每條記錄對應的Bean類：UserBehavior，和CSV記錄格式保持一致即可，表示時間的ts字段，使用了JsonFormat註解，在序列化的時候以此來控制格式：

<code>public class UserBehavior {

    @JsonFormat
    private long user_id;

    @JsonFormat
    private long item_id;

    @JsonFormat
    private long category_id;

    @JsonFormat
    private String behavior;

    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss'Z'")
    private Date ts;

    public UserBehavior() {
    }

    public UserBehavior(long user_id, long item_id, long category_id, String behavior, Date ts) {
        this.user_id = user_id;
        this.item_id = item_id;
        this.category_id = category_id;
        this.behavior = behavior;
        this.ts = ts;
    }
}/<code>

• Java對象序列化成JSON的序列化類：JsonSerializer

<code>public class JsonSerializer {

    private final ObjectMapper jsonMapper = new ObjectMapper();

    public String toJSONString(T r) {
        try {
            return jsonMapper.writeValueAsString(r);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Could not serialize record: " + r, e);
        }
    }

    public byte[] toJSONBytes(T r) {
        try {
            return jsonMapper.writeValueAsBytes(r);
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Could not serialize record: " + r, e);
        }
    }
}/<code>

• 向kafka發送消息的工具類：KafkaProducer

<code>public class KafkaProducer implements Consumer {

    private final String topic;
    private final org.apache.kafka.clients.producer.KafkaProducer producer;
    private final JsonSerializer serializer;

    public KafkaProducer(String kafkaTopic, String kafkaBrokers) {
        this.topic = kafkaTopic;
        this.producer = new org.apache.kafka.clients.producer.KafkaProducer<>(createKafkaProperties(kafkaBrokers));
        this.serializer = new JsonSerializer<>();
    }

    @Override
    public void accept(UserBehavior record) {
        // 將對象序列化成byte數組
        byte[] data = serializer.toJSONBytes(record);
        // 封裝
        ProducerRecord kafkaRecord = new ProducerRecord 
<>(topic, data);
        // 發送
        producer.send(kafkaRecord);

        // 通過sleep控制消息的速度，請依據自身kafka配置以及flink服務器配置來調整
        try {
            Thread.sleep(500);
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }

    /**
     * kafka配置
     * @param brokers The brokers to connect to.
     * @return A Kafka producer configuration.
     */
    private static Properties createKafkaProperties(String brokers) {
        Properties kafkaProps = new Properties();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
        return kafkaProps;
    }
}/<code>

• 最後是應用類SendMessageApplication，CSV文件路徑、kafka的topic和borker地址都在此設置，另外借助java8的Stream API，只需少量代碼即可完成所有工作：

<code>public class SendMessageApplication {

    public static void main(String[] args) throws Exception {
        // 文件地址
        String filePath = "D:\\temp\\202005\\02\\UserBehavior.csv";
        // kafka topic
        String topic = "user_behavior";
        // kafka borker地址
        String broker = "192.168.50.43:9092";


        Stream.generate(new UserBehaviorCsvFileReader(filePath))
                .sequential()
                .forEachOrdered(new KafkaProducer(topic, broker));
    }
}/<code>

驗證

• 請確保kafka已經就緒，並且名為user_behavior的topic已經創建；
• 請將CSV文件準備好；
• 確認SendMessageApplication.java中的文件地址、kafka topic、kafka broker三個參數準確無誤；
• 運行SendMessageApplication.java；
• 開啟一個 控制檯消息kafka消息，參考命令如下：

<code>./kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 \
--topic user_behavior \
--consumer-property group.id=old-consumer-test \
--consumer-property consumer.id=old-consumer-cl \
--from-beginning/<code>

正常情況下可以立即見到消息，如下圖：

至此，通過Java應用模擬用戶行為消息流的操作就完成了，接下來的flink實戰就用這個作為數據源；

歡迎關注我的公眾號：程序員欣宸