2. 程式人生 > 程式設計 >Java將CSV的資料傳送到kafka的示例

Java將CSV的資料傳送到kafka的示例

阿新 • • 發佈：2020-11-17

為什麼將CSV的資料發到kafka

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

Java將CSV的資料傳送到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）

關於資料集

  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')
時間戳 行為發生的時間戳
時間字串 根據時間戳欄位生成的時間字串

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

Java應用簡介

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

  1. 從CSV讀取記錄的工具類：UserBehaviorCsvFileReader
  2. 每條記錄對應的Bean類：UserBehavior
  3. Java物件序列化成JSON的序列化類：JsonSerializer
  4. 向kafka傳送訊息的工具類：KafkaProducer
  5. 應用類，程式入口：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這個資料夾下，如下圖紅框所示：

Java將CSV的資料傳送到kafka的示例

編碼

建立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 = 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;
 }
}

最後是應用類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
  • 正常情況下可以立即見到訊息，如下圖：

Java將CSV的資料傳送到kafka的示例

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

以上就是Java將CSV的資料傳送到kafka得示例的詳細內容，更多關於Java CSV的資料傳送到kafka的資料請關注我們其它相關文章！

相關推薦

Java將CSV的資料傳送到kafka的示例

為什麼將CSV的資料發到kafka flink做流式計算時，選用kafka訊息作為資料來源是常用手段，因此在學習和開發flink過程中，也會將資料集檔案中的記錄傳送到kafka，來模擬不間斷資料；

java處理csv檔案上傳示例詳解

前言：示例只是做了一個最最基礎的上傳csv的示例，如果要引用到程式碼中去，還需要根據自己的業務自行新增一些邏輯處理。

JAVA------UDP接受資料傳送資料

技術標籤：網路javaUDP網路協議 網路程式設計三要素： IP地址埠協議 一、IP地址

記錄一下遇到的問題 java將json資料解析為sql語句

這樣的json資料轉換成sql語句 解決辦法 import java.util.Iterator; import java.util.Set; import java.util.Map.Entry;

頁面關閉，也能將統計資料傳送到Web 伺服器

頁面關閉（前/時/後）將統計資料傳送到 Web 伺服器的方案： 一、XMLHttpRequest 缺點：定時器（setInterval）間隔時間不好把握，Ajax 也會浪費大量的資源。

將CSV的資料傳送到kafka(java版)

歡迎訪問我的GitHub https://github.com/zq2599/blog_demos 內容：所有原創文章分類彙總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；

利用pandas向一個csv檔案追加寫入資料的實現示例

我們越來越多的使用pandas進行資料處理，有時需要向一個已經存在的csv檔案寫入資料，傳統的方法之前我也有些過，向txt，excel檔案寫入資料，傳送門：Python將二維列表（list）的資料輸出（TXT，Excel）

java利用udp實現傳送資料

本文例項為大家分享了java利用udp實現傳送資料的具體程式碼，供大家參考，具體內容如下

Logback將日誌輸出到Kafka配置示例

1.maven依賴 <dependency> <groupId>com.github.danielwegener</groupId> <artifactId>logback-kafka-appender</artifactId>

java 將資料載入到記憶體中的操作

將資料載入到記憶體中 1、建立InitListener.java package app.util; import java.util.HashMap; import java.util.Map;

idea設定：將普通資料夾設定為resourses資原始檔夾或者java程式碼根目錄資料夾

如下圖所示的resources資料夾不是一個資原始檔夾，所以無法讀取資料夾下配置檔案的內容，配置檔案也無法生效。具體操作如下：

Java將資料生成XML檔案並進行壓縮成GZ格式

技術標籤：【XML】javaxmlgzipxstream 近期和PC端對接介面，在線上環境經常出現PC端介面拉取資料時後臺負載高的情況，為了解決這個問題，我們將介面轉換成xml檔案格式，每當PC客戶端啟動時會拉取服務端最新的xml

java 將查詢到的值 存下_javaSE第十四部分 JDBC(3)java查詢mysql資料並將結果集封裝為JavaBean陣列...

技術標籤：java 將查詢到的值 存下 將sql的查詢結果封裝為JavaBean 假如有一個表emp

safegraph資料預處理（二）：將指定資料夾內的.csv檔案合併

技術標籤：safegraphpython 將weekly-pattern-2020-12-14-backfill中的，指定資料夾內（當前層）的xxx-partx.csv合併。

safegraph資料預處理（三）：將csv檔案按指定欄位不同的值進行拆分

技術標籤：safegraphpython 將Nin1.csv按region不同的值進行拆分，儲存為xxx-region.csv，經驗證全部55個子檔案大小之和等於父檔案的大小。

Java 七行程式碼搞定將excel資料匯入到DB

技術標籤：ORMJavaBee import java.util.List; import org.junit.Test; import org.teasoft.bee.osql.SuidRich;

用Java實現將多級資料夾下的所有檔案統一放到一個資料夾中

package USEif; import java.io.File;import java.io.FileInputStream;import java.io.FileOutputStream;import java.io.IOException;

Hudi-Flink SQL實時讀取Hudi表資料 Hudi-Flink消費kafka將增量資料實時寫入Hudi

程式碼如下（hudi表實時寫入參考上一篇[Hudi-Flink消費kafka將增量資料實時寫入Hudi]）

通過canal+kafka將mysql資料匯入StarRocks

 背景 在支援客戶中，我們發現有一些客戶公司已經存在一些資料通道，不允許業務直接消費MySQL Binlog，所有的資料消費都是從Kafka中獲取，所以寫這篇文件分享下如何消費Kafka中canal格式的資料寫到到starrocks，實現

如何將 Text, XML, CSV 資料檔案匯入 MySQL

原文出處： freenik@Jianshu 將 外部資料匯入(import)資料庫是在資料庫應用中一個很常見的需求。其實這就是在資料的管理和操作中的ETL (Extract,transform, load)的L(Load)部分，也就是說，將特定結構(structure)或