Java将CSV的数据发送到kafka的示例

为什么将CSV的数据发到kafka

  • flink做流式计算时,选用kafka消息作为数据源是常用手段,因此在学习和开发flink过程中,也会将数据集文件中的记录发送到kafka,来模拟不间断数据;
  • 整个流程如下:

Java将CSV的数据发送到kafka的示例_第1张图片

  • 您可能会觉得这样做多此一举: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) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址,ssh协议

这个git项目中有多个文件夹,本章源码在flinksql这个文件夹下,如下图红框所示:

Java将CSV的数据发送到kafka的示例_第2张图片

编码

创建maven工程,pom.xml如下,比较重要的jackson和javacsv的依赖:



 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:

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;
 }
}

最后是应用类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应用模拟用户行为消息流的操作就完成了,接下来的flink实战就用这个作为数据源;

以上就是Java将CSV的数据发送到kafka得示例的详细内容,更多关于Java CSV的数据发送到kafka的资料请关注脚本之家其它相关文章!

你可能感兴趣的