將CSV的數據發送到kafka(java版)

歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；

為什么將CSV的數據發到kafka

1. flink做流式計算時，選用kafka消息作為數據源是常用手段，因此在學習和開發flink過程中，也會將數據集文件中的記錄發送到kafka，來模擬不間斷數據；
2. 整個流程如下：
   
3. 您可能會覺得這樣做多此一舉：flink直接讀取CSV不就行了嗎？這樣做的原因如下：
4. 首先，這是學習和開發時的做法，數據集是CSV文件，而生產環境的實時數據卻是kafka數據源；
5. 其次，Java應用中可以加入一些特殊邏輯，例如數據處理，匯總統計（用來和flink結果對比驗證）；
6. 另外，如果兩條記錄實際的間隔時間如果是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）

關於數據集

1. 本次實戰用到的數據集是CSV文件，里面是一百零四萬條淘寶用戶行為數據，該數據來源是阿里雲天池公開數據集，我對此數據做了少量調整；
2. 此CSV文件可以在CSDN下載，地址：https://download.csdn.net/download/boling_cavalry/12381698
3. 也可以在我的Github下載，地址：https://raw.githubusercontent.com/zq2599/blog_demos/master/files/UserBehavior.7z
4. 該CSV文件的內容，一共有六列，每列的含義如下表：

列名稱	說明
用戶ID	整數類型，序列化后的用戶ID
商品ID	整數類型，序列化后的商品ID
商品類目ID	整數類型，序列化后的商品所屬類目ID
行為類型	字符串，枚舉類型，包括('pv', 'buy', 'cart', 'fav')
時間戳	行為發生的時間戳
時間字符串	根據時間戳字段生成的時間字符串

5. 關於該數據集的詳情，請參考《准備數據集用於flink學習》

Java應用簡介

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

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

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

直接下載源碼

1. 如果您不想寫代碼，您可以直接從GitHub下載這個工程的源碼，地址和鏈接信息如下表所示：

名稱	鏈接	備注
項目主頁	https://github.com/zq2599/blog_demos	該項目在GitHub上的主頁
git倉庫地址(https)	https://github.com/zq2599/blog_demos.git	該項目源碼的倉庫地址，https協議
git倉庫地址(ssh)	[e-mail]:zq2599/blog_demos.git	該項目源碼的倉庫地址，ssh協議

2. 這個git項目中有多個文件夾，本章源碼在flinksql這個文件夾下，如下圖紅框所示：
   

編碼

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

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.bolingcavalry</groupId>
    <artifactId>flinksql</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.10.0</flink.version>
        <kafka.version>2.2.0</kafka.version>
        <java.version>1.8</java.version>
        <scala.binary.version>2.11</scala.binary.version>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>

        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.10.1</version>
        </dependency>

        <!-- Logging dependencies -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>net.sourceforge.javacsv</groupId>
            <artifactId>javacsv</artifactId>
            <version>2.0</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <!-- Java Compiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>

            <!-- Shade plugin to include all dependencies -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <!-- Run shade goal on package phase -->
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

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

public class UserBehaviorCsvFileReader implements Supplier<UserBehavior> {

    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;
    }
}

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

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;
    }
}

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

public class JsonSerializer<T> {

    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);
        }
    }
}

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

public class KafkaProducer implements Consumer<UserBehavior> {

    private final String topic;
    private final org.apache.kafka.clients.producer.KafkaProducer<byte[], byte[]> producer;
    private final JsonSerializer<UserBehavior> 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<byte[], byte[]> 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;
    }
}

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

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));
    }
}

驗證

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

./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

6. 正常情況下可以立即見到消息，如下圖：
   
   至此，通過Java應用模擬用戶行為消息流的操作就完成了，接下來的flink實戰就用這個作為數據源；

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

微信搜索「程序員欣宸」，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos