將CSV的數據發送到kafka

為什麼將CSV的數據發到kafka

如何將CSV的數據發送到kafka

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

版本信息

關於數據集

https://download.csdn.net/download/boling_cavalry/12381698

https://raw.githubusercontent.com/zq2599/blog_demos/master/files/UserBehavior.7z

Java應用簡介

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

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

直接下載源碼

編碼

<code> 4.0.0 com.bolingcavalry flinksql 1.0-SNAPSHOT UTF-8 1.10.0 2.2.0 1.8 2.11 ${java.version} ${java.version} org.apache.kafka kafka-clients ${kafka.version} com.fasterxml.jackson.core jackson-databind 2.9.10.1 org.slf4j slf4j-log4j12 1.7.7 runtime log4j log4j 1.2.17 runtime net.sourceforge.javacsv javacsv 2.0 org.apache.maven.plugins maven-compiler-plugin 3.1 ${java.version} ${java.version} org.apache.maven.plugins maven-shade-plugin 3.0.0 package shade *:* META-INF/*.SF META-INF/*.DSA META-INF/*.RSA /<code>從CSV讀取記錄的工具類：UserBehaviorCsvFileReader，後面在主程序中會用到java8的Steam API來處理集合，所以UserBehaviorCsvFileReader實現了Supplier接口：

<code>public class UserBehaviorCsvFileReader implements Supplier { 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; } }/<code>每條記錄對應的Bean類：UserBehavior，和CSV記錄格式保持一致即可，表示時間的ts字段，使用了JsonFormat註解，在序列化的時候以此來控制格式：

<code>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; } }/<code>Java對象序列化成JSON的序列化類：JsonSerializer

<code>public class JsonSerializer { 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); } } }/<code>向kafka發送消息的工具類：KafkaProducer

<code>public class KafkaProducer implements Consumer { private final String topic; private final org.apache.kafka.clients.producer.KafkaProducer producer; private final JsonSerializer 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 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; } }/<code>最後是應用類SendMessageApplication，CSV文件路徑、kafka的topic和borker地址都在此設置，另外借助java8的Stream API，只需少量代碼即可完成所有工作：

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

驗證

<code>./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/<code>正常情況下可以立即見到消息，如下圖：

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

歡迎關注我的公眾號：程序員欣宸