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

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

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

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

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

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

如何將CSV的數據發(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）

關于數據集

本次實戰(zhàn)用到的數據集是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')
時間戳	行為發(fā)生的時間戳
時間字符串	根據時間戳字段生成的時間字符串

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

Java應用簡介

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

從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協議
git倉庫地址(ssh)	git@github.com:zq2599/blog_demos.git	該項目源碼的倉庫地址，ssh協議

這個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實現了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數組
  byte[] data = serializer.toJSONBytes(record);
  // 封裝
  ProducerRecord<byte[], byte[]> kafkaRecord = new ProducerRecord<>(topic, data);
  // 發(fā)送
  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已經創(chuàng)建；
請將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實戰(zhàn)就用這個作為數據源；

以上就是Java將CSV的數據發(fā)送到kafka得示例的詳細內容，更多關于Java CSV的數據發(fā)送到kafka的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

Java volatile關鍵字原理剖析與實例講解
volatile是Java提供的一種輕量級的同步機制，Java?語言包含兩種內在的同步機制：同步塊（或方法）和?volatile?變量，本文將詳細為大家總結Java volatile關鍵字，通過詳細的代碼示例給大家介紹的非常詳細,需要的朋友可以參考下
2023-07-07
Java Socket編程實例（一）- TCP基本使用
這篇文章主要講解Java Socket編程中TCP的基本使用，希望能給大家做一個參考。
2016-06-06
Java?C++題解eetcode940不同的子序列?II
這篇文章主要為大家介紹了Java?C++題解eetcode940不同的子序列?II實現示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2022-10-10
Java用三元運算符判斷奇數和偶數的簡單實現
這篇文章主要介紹了Java用三元運算符判斷奇數和偶數的簡單實現,需要的朋友可以參考下
2014-02-02
Java實現DES加密與解密，md5加密以及Java實現MD5加密解密類
這篇文章主要介紹了Java實現DES加密與解密，md5加密以及Java實現MD5加密解密類 ,需要的朋友可以參考下
2015-11-11
實例講解Java設計模式編程中如何運用代理模式
這篇文章主要介紹了Java設計模式編程中如何運用代理模式,文中舉了普通代理和強制代理的例子作為代理模式的擴展內容,需要的朋友可以參考下
2016-02-02
springboot各種下載文件的方式匯總
下載功能其實就是用戶輸入指定文件路徑信息,然后把文件返回給用戶,下面這篇文章主要給大家介紹了關于springboot各種下載文件的方式,需要的朋友可以參考下
2022-10-10
java簡單解析xls文件的方法示例【讀取和寫入】
這篇文章主要介紹了java簡單解析xls文件的方法,結合實例形式分析了java針對xls文件的讀取和寫入相關操作技巧與注意事項,需要的朋友可以參考下
2017-06-06
java編程題之從上往下打印出二叉樹
這篇文章主要為大家詳細介紹了java編程題之從上往下打印出二叉樹，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2019-03-03
Java詳解表格的創(chuàng)建與使用流程
這篇文章主要介紹了怎么用Java來創(chuàng)建和使用表格，表格是我們經常要用的工具，但是你有想過自己怎么去實現它嗎，感興趣的朋友跟隨文章往下看看吧
2022-04-04