Java將CSV的數(shù)據(jù)發(fā)送到kafka的示例

更新時間：2020年11月16日 09:54:10 作者：zq2599

這篇文章主要介紹了Java將CSV的數(shù)據(jù)發(fā)送到kafka得示例，幫助大家更好得理解和使用Java，感興趣的朋友可以了解下

為什么將CSV的數(shù)據(jù)發(fā)到kafka

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

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

如何將CSV的數(shù)據(jù)發(fā)送到kafka

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

版本信息

JDK：1.8.0_181
開發(fā)工具：IntelliJ IDEA 2019.2.1 (Ultimate Edition)
開發(fā)環(huán)境：Win10
Zookeeper：3.4.13
Kafka：2.4.0（scala：2.12）

關于數(shù)據(jù)集

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

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

關于該數(shù)據(jù)集的詳情，請參考《準備數(shù)據(jù)集用于flink學習》

Java應用簡介

編碼前，先把具體內(nèi)容列出來，然后再挨個實現(xiàn)：

從CSV讀取記錄的工具類：UserBehaviorCsvFileReader
每條記錄對應的Bean類：UserBehavior
Java對象序列化成JSON的序列化類：JsonSerializer
向kafka發(fā)送消息的工具類：KafkaProducer
應用類，程序入口：SendMessageApplication

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

直接下載源碼

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

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

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

編碼

創(chuàng)建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實現(xiàn)了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發(fā)送消息的工具類：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數(shù)組
  byte[] data = serializer.toJSONBytes(record);
  // 封裝
  ProducerRecord<byte[], byte[]> kafkaRecord = new ProducerRecord<>(topic, data);
  // 發(fā)送
  producer.send(kafkaRecord);

  // 通過sleep控制消息的速度，請依據(jù)自身kafka配置以及flink服務器配置來調(diào)整
  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已經(jīng)就緒，并且名為user_behavior的topic已經(jīng)創(chuàng)建；
請將CSV文件準備好；
確認SendMessageApplication.java中的文件地址、kafka topic、kafka broker三個參數(shù)準確無誤；
運行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