將CSV的資料發送到kafka(java版)-有解無憂

歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

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

為什么將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應用，并驗證；

版本資訊

JDK：1.8.0_181
開發工具：IntelliJ IDEA 2019.2.1 (Ultimate Edition)
開發環境：Win10
Zookeeper：3.4.13
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檔案的內容，一共有六列，每列的含義如下表：

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

關于該資料集的詳情，請參考《準備資料集用于flink學習》

Java應用簡介

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

從CSV讀取記錄的工具類：UserBehaviorCsvFileReader
每條記錄對應的Bean類：UserBehavior
Java物件序列化成JSON的序列化類：JsonSerializer
向kafka發送訊息的工具類：KafkaProducer
應用類，程式入口：SendMessageApplication

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

直接下載原始碼

如果您不想寫代碼，您可以直接從GitHub下載這個工程的原始碼，地址和鏈接資訊如下表所示：

名稱	鏈接	備注
專案主頁	https://github.com/zq2599/blog_demos	該專案在GitHub上的主頁
git倉庫地址(https)	https://github.com/zq2599/blog_demos.git	該專案原始碼的倉庫地址，https協議
git倉庫地址(ssh)	[email protected]:zq2599/blog_demos.git	該專案原始碼的倉庫地址，ssh協議

這個git專案中有多個檔案夾，本章原始碼在flinksql這個檔案夾下，如下圖紅框所示：

編碼

創建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>

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

每條記錄對應的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;
    }
}

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

向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 = https://www.cnblogs.com/bolingcavalry/archive/2020/11/16/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;
    }
}

最后是應用類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));
    }
}

驗證

請確保kafka已經就緒，并且名為user_behavior的topic已經創建；
請將CSV檔案準備好；
確認SendMessageApplication.java中的檔案地址、kafka topic、kafka broker三個引數準確無誤；
運行SendMessageApplication.java；
開啟一個控制臺訊息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

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

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

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

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

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/219389.html

標籤：其他

上一篇：Django-模板

下一篇：JAVA中常見的阻塞佇列詳解