### 歡迎訪問我的GitHub [https://github.com/zq2599/blog_demos](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學習》](https://blog.csdn.net/boling_cavalry/article/details/106033059) ### 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)| [email protected]:zq2599/blog_demos.git | 該專案原始碼的倉庫地址，ssh協議 | 2. 這個git專案中有多個資料夾，本章原始碼在flinksql這個資料夾下，如下圖紅框所示：  ### 編碼 1. 建立maven工程，pom.xml如下，比較重要的jackson和javacsv的依賴： ```xml ``` 2. 從CSV讀取記錄的工具類：UserBehaviorCsvFileReader，後面在主程式中會用到java8的Steam API來處理集合，所以UserBehaviorCsvFileReader實現了Supplier介面： ```java public class UserBehaviorCsvFileReader implements