为什么将CSV的数据发到kafka
如何将CSV的数据发送到kafka
前面的图可以看出，读取CSV再发送消息到kafka的操作是Java应用所为，因此今天的主要工作就是开发这个Java应用，并验证；
版本信息
关于数据集
| 列名称 | 说明 |
|---|---|
| 用户ID | 整数类型，序列化后的用户ID |
| 商品ID | 整数类型，序列化后的商品ID |
| 商品类目ID | 整数类型，序列化后的商品所属类目ID |
| 行为类型 | 字符串，枚举类型，包括('pv', 'buy', 'cart', 'fav') |
| 时间戳 | 行为发生的时间戳 |
| 时间字符串 | 根据时间戳字段生成的时间字符串 |
Java应用简介
编码前，先把具体内容列出来，然后再挨个实现：
上述五个类即可完成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这个文件夹下，如下图红框所示：
编码
创建maven工程，pom.xml如下，比较重要的jackson和javacsv的依赖：
<?xml version="1.0" encoding="UTF-8"?>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
从CSV读取记录的工具类：UserBehaviorCsvFileReader，后面在主程序中会用到java8的Steam API来处理集合，所以UserBehaviorCsvFileReader实现了Supplier接口：
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; } }
每条记录对应的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{ 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：
京公网安备 11010802040832号