为什么将CSV的数据发到kafka
- flink做流式计算时，选用kafka消息作为数据源是常用手段，因此在学习和开发flink过程中，也会将数据集文件中的记录发送到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）
关于数据集
- 本次实战用到的数据集是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') |
| 时间戳 | 行为发生的时间戳 |
| 时间字符串 | 根据时间戳字段生成的时间字符串 |
- 关于该数据集的详情，请参考《准备数据集用于flink学习》
Java应用简介
编码前，先把具体内容列出来，然后再挨个实现：
- 从CSV读取记录的工具类：UserBehaviorCsvFileReader
- 每条记录对应的Bean类：UserBehavior
- Java对象序列化成JSON的序列化类：JsonSerializer
- 向kafka发送消息的工具类：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这个文件夹下，如下图红框所示：
编码
创建maven工程，pom.xml如下，比较重要的jackson和javacsv的依赖：
4.0.0 com.bolingcavalry flinksql1.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-databind2.9.10.1 org.slf4j slf4j-log4j121.7.7 runtime log4j log4j1.2.17 runtime net.sourceforge.javacsv javacsv2.0 org.apache.maven.plugins maven-compiler-plugin3.1 ${java.version} ${java.version} org.apache.maven.plugins maven-shade-plugin3.0.0 package shade *:* meta-INF 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已经创建；
- 请将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实战就用这个作为数据源；
以上就是Java将CSV的数据发送到kafka得示例的详细内容，更多关于Java CSV的数据发送到kafka的资料请关注考高分网其它相关文章！
