图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flink › 总结型› Flink实现自定义Avro序列化(Source/Sink)到kafka
分享

Flink实现自定义Avro序列化(Source/Sink)到kafka

levycui 2021-8-10 17:55:44 发表于 总结型 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 4225
问题导读：
1、Avro提供的技术支持包括哪些？
2、Avro优点有哪些？
3、如何使用Java自定义序列化到kafka？
4、Flink如何实现Avro自定义序列化到Kafka？

前言

最近一直在研究如果提高kafka中读取效率，之前一直使用字符串的方式将数据写入到kafka中。当数据将特别大的时候发现效率不是很好，偶然之间接触到了Avro序列化，发现kafka也是支持Avro的方式于是就有了本篇文章。

环境所依赖的pom文件

  1. <dependencies>

  2.         <dependency>

  3.             <groupId>org.apache.avro</groupId>

  4.             <artifactId>avro</artifactId>

  5.             <version>1.8.2</version>

  6.         </dependency>

  7.         <dependency>

  8.             <groupId>org.apache.flink</groupId>

  9.             <artifactId>flink-scala_2.12</artifactId>

  10.             <version>1.10.1</version>

  11.         </dependency>

  12.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->

  13.         <dependency>

  14.             <groupId>org.apache.flink</groupId>

  15.             <artifactId>flink-streaming-scala_2.12</artifactId>

  16.             <version>1.10.1</version>

  17.         </dependency>

  18.         <dependency>

  19.             <groupId>org.apache.flink</groupId>

  20.             <artifactId>flink-connector-kafka-0.11_2.12</artifactId>

  21.             <version>1.10.1</version>

  22.         </dependency>

  23.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-avro -->

  24.         <dependency>

  25.             <groupId>org.apache.flink</groupId>

  26.             <artifactId>flink-avro</artifactId>

  27.             <version>1.10.1</version>

  28.         </dependency>

  29.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->

  30.         <dependency>

  31.             <groupId>org.apache.kafka</groupId>

  32.             <artifactId>kafka-clients</artifactId>

  33.             <version>1.0.0</version>

  34.         </dependency>

  35.         <dependency>

  36.             <groupId>org.apache.kafka</groupId>

  37.             <artifactId>kafka-streams</artifactId>

  38.             <version>1.0.0</version>

  39.         </dependency>

  40.     </dependencies>

  41.     <build>

  42.         <plugins>

  43.             <plugin>

  44.                 <groupId>org.apache.avro</groupId>

  45.                 <artifactId>avro-maven-plugin</artifactId>

  46.                 <version>1.8.2</version>

  47.                 <executions>

  48.                     <execution>

  49.                         <phase>generate-sources</phase>

  50.                         <goals>

  51.                             <goal>schema</goal>

  52.                         </goals>

  53.                         <configuration>

  54.                             <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>

  55.                             <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>

  56.                         </configuration>

  57.                     </execution>

  58.                 </executions>

  59.             </plugin>

  60.             <plugin>

  61.                 <groupId>org.apache.maven.plugins</groupId>

  62.                 <artifactId>maven-compiler-plugin</artifactId>

  63.                 <configuration>

  64.                     <source>1.6</source>

  65.                     <target>1.6</target>

  66.                 </configuration>

  67.             </plugin>

  68.         </plugins>

  69.     </build>
复制代码


一、Avro提供的技术支持包括以下五个方面：

  •     优秀的数据结构；
  •     一个紧凑的，快速的，二进制数据格式；
  •     一个容器文件，用来存储持久化数据；
  •     RPC远程过程调用；
  •     集成最简单的动态语言。读取或者写入数据文件，使用或实现RPC协议均不需要代码实现。对于静态- - 语言编写的话需要实现；

二、Avro优点

  •     二进制消息，性能好/效率高
  •     使用JSON描述模式
  •     模式和数据统一存储，消息自描述，不需要生成stub代码（支持生成IDL）
  •     RPC调用在握手阶段交换模式定义
  •     包含完整的客户端/服务端堆栈，可快速实现RPC
  •     支持同步和异步通信
  •     支持动态消息
  •     模式定义允许定义数据的排序（序列化时会遵循这个顺序）
  •     提供了基于Jetty内核的服务基于Netty的服务

三、Avro Json格式介绍

  1. {

  2.     "namespace": "com.avro.bean",

  3.     "type": "record",

  4.     "name": "UserBehavior",

  5.     "fields": [

  6.         {"name": "userId", "type": "long"},

  7.         {"name": "itemId",  "type": "long"},

  8.         {"name": "categoryId", "type": "int"},

  9.         {"name": "behavior", "type": "string"},

  10.         {"name": "timestamp", "type": "long"}

  11.     ]

  12. }
复制代码

  •     namespace : 要生成的目录
  •     type ： 类型 avro 使用 record
  •     name : 会自动生成对应的对象
  •     fields : 要指定的字段

注意: 创建的文件后缀名一定要叫 avsc

我们使用idea 生成 UserBehavior 对象
20210108195633291.gif

四、使用Java自定义序列化到kafka

         首先我们先使用 Java编写Kafka客户端写入数据和消费数据。

4.1 准备测试数据

  1.     543462,1715,1464116,pv,1511658000

  2.     662867,2244074,1575622,pv,1511658000

  3.     561558,3611281,965809,pv,1511658000

  4.     894923,3076029,1879194,pv,1511658000

  5.     834377,4541270,3738615,pv,1511658000

  6.     315321,942195,4339722,pv,1511658000

  7.     625915,1162383,570735,pv,1511658000
复制代码



4.2 自定义Avro 序列化和反序列化

首先我们需要实现2个类分别为Serializer和Deserializer分别是序列化和反序列化

  1. package com.avro.AvroUtil;

  2. 

  3. import com.avro.bean.UserBehavior;

  4. import org.apache.avro.io.BinaryDecoder;

  5. import org.apache.avro.io.BinaryEncoder;

  6. import org.apache.avro.io.DecoderFactory;

  7. import org.apache.avro.io.EncoderFactory;

  8. import org.apache.avro.specific.SpecificDatumReader;

  9. import org.apache.avro.specific.SpecificDatumWriter;

  10. import org.apache.kafka.common.serialization.Deserializer;

  11. import org.apache.kafka.common.serialization.Serializer;

  12. 

  13. import java.io.ByteArrayInputStream;

  14. import java.io.ByteArrayOutputStream;

  15. import java.io.IOException;

  16. import java.util.Map;

  17. 

  18. /**

  19.  * @author 大数据老哥

  20.  * @version V1.0

  21.  * @Package com.avro.AvroUtil

  22.  * @File ：SimpleAvroSchemaJava.java

  23.  * @date 2021/1/8 20:02

  24.  */

  25. /**

  26.  *  自定义序列化和反序列化

  27.  */

  28. public class SimpleAvroSchemaJava implements Serializer<UserBehavior>, Deserializer<UserBehavior> {

  29.     

  30.     @Override

  31.     public void configure(Map<String, ?> map, boolean b) {

  32. 

  33.     }

  34.     //序列化方法

  35.     @Override

  36.     public byte[] serialize(String s, UserBehavior userBehavior) {

  37.         // 创建序列化执行器

  38.         SpecificDatumWriter<UserBehavior> writer = new SpecificDatumWriter<UserBehavior>(userBehavior.getSchema());

  39.          // 创建一个流 用存储序列化后的二进制文件

  40.         ByteArrayOutputStream out = new ByteArrayOutputStream();

  41.         // 创建二进制编码器

  42.         BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(out, null);

  43.         try {

  44.             // 数据入都流中

  45.             writer.write(userBehavior, encoder);

  46.         } catch (IOException e) {

  47.             e.printStackTrace();

  48.         }

  49. 

  50.         return out.toByteArray();

  51.     }

  52. 

  53.     @Override

  54.     public void close() {

  55. 

  56.     }

  57. 

  58.     //反序列化

  59.     @Override

  60.     public UserBehavior deserialize(String s, byte[] bytes) {

  61.         // 用来保存结果数据

  62.         UserBehavior userBehavior = new UserBehavior();

  63.         // 创建输入流用来读取二进制文件

  64.         ByteArrayInputStream arrayInputStream = new ByteArrayInputStream(bytes);

  65.         // 创建输入序列化执行器

  66.         SpecificDatumReader<UserBehavior> stockSpecificDatumReader = new SpecificDatumReader<UserBehavior>(userBehavior.getSchema());

  67.         //创建二进制解码器

  68.         BinaryDecoder binaryDecoder = DecoderFactory.get().directBinaryDecoder(arrayInputStream, null);

  69.         try {

  70.             // 数据读取

  71.             userBehavior=stockSpecificDatumReader.read(null, binaryDecoder);

  72.         } catch (IOException e) {

  73.             e.printStackTrace();

  74.         }

  75.         // 结果返回

  76.         return userBehavior;

  77.     }

  78. }

复制代码

4.3 创建序列化对象

  1. package com.avro.kafka;

  2. import com.avro.bean.UserBehavior;

  3. import org.apache.kafka.clients.producer.KafkaProducer;

  4. import org.apache.kafka.clients.producer.ProducerRecord;

  5. import java.io.BufferedReader;

  6. import java.io.FileReader;

  7. import java.util.ArrayList;

  8. import java.util.List;

  9. import java.util.Properties;

  10. 

  11. /**

  12.  * @author 大数据老哥

  13.  * @version V1.0

  14.  * @Package com.avro.kafka

  15.  * @File ：UserBehaviorProducerKafka.java

  16.  * @date 2021/1/8 20:14

  17.  */

  18. 

  19. public class UserBehaviorProducerKafka {

  20.     public static void main(String[] args) throws InterruptedException {

  21.         // 获取数据

  22.         List<UserBehavior> data = getData();

  23.         // 创建配置文件

  24.         Properties props = new Properties();

  25.         props.setProperty("bootstrap.servers", "192.168.100.201:9092,192.168.100.202:9092,192.168.100.203:9092");

  26.         props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

  27.         props.setProperty("value.serializer", "com.avro.AvroUtil.SimpleAvroSchemaJava");

  28.         // 创建kafka的生产者

  29.         KafkaProducer<String, UserBehavior> userBehaviorProducer = new KafkaProducer<String, UserBehavior>(props);

  30.         // 循环遍历数据

  31.         for (UserBehavior userBehavior : data) {

  32.             ProducerRecord<String, UserBehavior> producerRecord = new ProducerRecord<String, UserBehavior>("UserBehaviorKafka", userBehavior);

  33.             userBehaviorProducer.send(producerRecord);

  34.             System.out.println("数据写入成功"+data);

  35.             Thread.sleep(1000);

  36.         }

  37.     }

  38. 

  39.     public static List<UserBehavior> getData() {

  40.         ArrayList<UserBehavior> userBehaviors = new ArrayList<UserBehavior>();

  41.         try {

  42.             BufferedReader br = new BufferedReader(new FileReader(new File("data/UserBehavior.csv")));

  43.             String line = "";

  44.             while ((line = br.readLine()) != null) {

  45.                 String[] split = line.split(",");

  46.              userBehaviors.add( new UserBehavior(Long.parseLong(split[0]), Long.parseLong(split[1]), Integer.parseInt(split[2]), split[3], Long.parseLong(split[4])));

  47.             }

  48.         } catch (Exception e) {

  49.             e.printStackTrace();

  50.         }

  51.         return userBehaviors;

  52.     }

  53. }

复制代码


注意：value.serializer 一定要指定我们自己写好的那个反序列化类，负责会无效

4.4 创建反序列化对象

  1. package com.avro.kafka;

  2. import com.avro.bean.UserBehavior;

  3. import org.apache.kafka.clients.consumer.ConsumerRecord;

  4. import org.apache.kafka.clients.consumer.ConsumerRecords;

  5. import org.apache.kafka.clients.consumer.KafkaConsumer;

  6. import java.util.Arrays;

  7. import java.util.Properties;

  8. 

  9. /**

  10.  * @author 大数据老哥

  11.  * @version V1.0

  12.  * @Package com.avro.kafka

  13.  * @File ：UserBehaviorConsumer.java

  14.  * @date 2021/1/8 20:58

  15.  */

  16. public class UserBehaviorConsumer {

  17. 

  18.     public static void main(String[] args) {

  19.         Properties prop = new Properties();

  20.         prop.put("bootstrap.servers", "192.168.100.201:9092,192.168.100.202:9092,192.168.100.203:9092");

  21.         prop.put("group.id", "UserBehavior");

  22.         prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

  23.         // 设置反序列化类为自定义的avro反序列化类

  24.         prop.put("value.deserializer", "com.avro.AvroUtil.SimpleAvroSchemaJava");

  25.         KafkaConsumer<String, UserBehavior> consumer = new KafkaConsumer<String, UserBehavior>(prop);

  26.         consumer.subscribe(Arrays.asList("UserBehaviorKafka"));

  27.         while (true) {

  28.             ConsumerRecords<String, UserBehavior> poll = consumer.poll(1000);

  29.             for (ConsumerRecord<String, UserBehavior> stringStockConsumerRecord : poll) {

  30.                 System.out.println(stringStockConsumerRecord.value());

  31.             }

  32.         }

  33.     }

  34. }
复制代码

4.5 启动运行

创建kafkaTopic 和启动一个消费者
  1. <div># 创建topic

  2. ./kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03:2181 --replication-factor 2 --partitions 3 --topic UserBehaviorKafka

  3. </div><div>

  4. </div><div># 模拟消费者

  5. ./kafka-console-consumer.sh --from-beginning --topic UserBehaviorKafka --zookeeper node01:2181,node02:2node03:2181</div>
复制代码


2021-08-10_173837.jpg

五、Flink 实现Avro自定义序列化到Kafka

到这里好多小伙们就说我Java实现了那Flink 不就改一下Consumer 和Producer 不就完了吗？

5.1 准备数据

  1.     543462,1715,1464116,pv,1511658000

  2.     662867,2244074,1575622,pv,1511658000

  3.     561558,3611281,965809,pv,1511658000

  4.     894923,3076029,1879194,pv,1511658000

  5.     834377,4541270,3738615,pv,1511658000

  6.     315321,942195,4339722,pv,1511658000

  7.     625915,1162383,570735,pv,1511658000
复制代码



5.2 创建Flink自定义Avro序列化和反序列化

当我们创建FlinkKafka连接器的时候发现使用Java那个类序列化发现不行，于是我们改为了系统自带的那个类进行测试。点击源码查看发系统自带的那个String其实实现的是DeserializationSchema和SerializationSchema，那我们是不是也可以模仿一个那？

2021-08-10_174027.jpg 

  1. <div>package com.avro.AvroUtil;

  2. 

  3. import com.avro.bean.UserBehavior;

  4. import com.typesafe.sslconfig.ssl.FakeChainedKeyStore;

  5. import org.apache.avro.io.BinaryDecoder;

  6. import org.apache.avro.io.BinaryEncoder;

  7. import org.apache.avro.io.DecoderFactory;

  8. import org.apache.avro.io.EncoderFactory;

  9. import org.apache.avro.specific.SpecificDatumReader;

  10. import org.apache.avro.specific.SpecificDatumWriter;

  11. import org.apache.flink.api.common.serialization.DeserializationSchema;

  12. import org.apache.flink.api.common.serialization.SerializationSchema;

  13. import org.apache.flink.api.common.typeinfo.TypeInformation;

  14. import org.apache.kafka.common.serialization.Deserializer;

  15. import org.apache.kafka.common.serialization.Serializer;

  16. 

  17. import java.io.ByteArrayInputStream;

  18. import java.io.ByteArrayOutputStream;

  19. import java.io.IOException;

  20. import java.util.Map;

  21. 

  22. /**

  23.  * @author 大数据老哥

  24.  * @version V1.0

  25.  * @Package com.avro.AvroUtil

  26.  * @File ：SimpleAvroSchemaFlink.java

  27.  * @date 2021/1/8 20:02

  28.  */

  29. 

  30. /**

  31.  *  自定义序列化和反序列化

  32.  */

  33. public class SimpleAvroSchemaFlink implements DeserializationSchema<UserBehavior>, SerializationSchema<UserBehavior> {

  34. 

  35.  

  36.     @Override

  37.     public byte[] serialize(UserBehavior userBehavior) {

  38.         // 创建序列化执行器

  39.         SpecificDatumWriter<UserBehavior> writer = new SpecificDatumWriter<UserBehavior>(userBehavior.getSchema());

  40.         // 创建一个流 用存储序列化后的二进制文件

  41.         ByteArrayOutputStream out = new ByteArrayOutputStream();

  42.         // 创建二进制编码器

  43.         BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(out, null);

  44.         try {

  45.             // 数据入都流中

  46.             writer.write(userBehavior, encoder);

  47.         } catch (IOException e) {

  48.             e.printStackTrace();

  49.         }

  50. 

  51.         return out.toByteArray();

  52.     }

  53. 

  54.     @Override

  55.     public TypeInformation<UserBehavior> getProducedType() {

  56.       return TypeInformation.of(UserBehavior.class);

  57.     }

  58. 

  59.     @Override

  60.     public UserBehavior deserialize(byte[] bytes) throws IOException {

  61.         // 用来保存结果数据

  62.         UserBehavior userBehavior = new UserBehavior();

  63.         // 创建输入流用来读取二进制文件

  64.         ByteArrayInputStream arrayInputStream = new ByteArrayInputStream(bytes);

  65.         // 创建输入序列化执行器

  66.         SpecificDatumReader<UserBehavior> stockSpecificDatumReader = new SpecificDatumReader<UserBehavior>(userBehavior.getSchema());

  67.         //创建二进制解码器

  68.         BinaryDecoder binaryDecoder = DecoderFactory.get().directBinaryDecoder(arrayInputStream, null);

  69.         try {

  70.             // 数据读取

  71.             userBehavior=stockSpecificDatumReader.read(null, binaryDecoder);

  72.         } catch (IOException e) {

  73.             e.printStackTrace();

  74.         }

  75.         // 结果返回

  76.         return userBehavior;

  77.     }

  78. 

  79.     @Override

  80.     public boolean isEndOfStream(UserBehavior userBehavior) {

  81.         return false;

  82.     }

  83. }

  84. </div>
复制代码



5.3 创建Flink Comsumer 反序列化

  1. package com.avro.FlinkKafka

  2. 

  3. import com.avro.AvroUtil.{SimpleAvroSchemaFlink}

  4. import com.avro.bean.UserBehavior

  5. import org.apache.flink.streaming.api.scala._

  6. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

  7. 

  8. import java.util.Properties

  9. 

  10. /**

  11.  * @Package com.avro.FlinkKafka

  12.  * @File ：UserBehaviorConsumerFlink.java

  13.  * @author 大数据老哥

  14.  * @date 2021/1/8 21:18

  15.  * @version V1.0

  16.  */

  17. object UserBehaviorConsumerFlink {

  18.   def main(args: Array[String]): Unit = {

  19.     //1.构建流处理运行环境

  20.     val env = StreamExecutionEnvironment.getExecutionEnvironment

  21.     env.setParallelism(1) // 设置并行度1 方便后面测试

  22.     // 2.设置kafka 配置信息

  23.     val prop = new Properties

  24.     prop.put("bootstrap.servers", "192.168.100.201:9092,192.168.100.202:9092,192.168.100.203:9092")

  25.     prop.put("group.id", "UserBehavior")

  26.     prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

  27.     // 设置反序列化类为自定义的avro反序列化类

  28.     prop.put("value.deserializer", "com.avro.AvroUtil.SimpleAvroSchemaFlink")

  29. 

  30.     //    val kafka: FlinkKafkaConsumer011[String] =  new FlinkKafkaConsumer011[String]("UserBehaviorKafka", new SimpleStringSchema(), prop)

  31.     // 3.构建Kafka 连接器

  32.     val kafka: FlinkKafkaConsumer011[UserBehavior] = new FlinkKafkaConsumer011[UserBehavior]("UserBehavior", new SimpleAvroSchemaFlink(), prop)

  33. 

  34.     //4.设置Flink层最新的数据开始消费

  35.     kafka.setStartFromLatest()

  36.     //5.基于kafka构建数据源

  37.     val data: DataStream[UserBehavior] = env.addSource(kafka)

  38.     //6.结果打印

  39.     data.print()

  40.     env.execute("UserBehaviorConsumerFlink")

  41.   }

  42. }

复制代码


5.4 创建Flink Producer 序列化

  1. package com.avro.FlinkKafka

  2. 

  3. import com.avro.AvroUtil.SimpleAvroSchemaFlink

  4. import com.avro.bean.UserBehavior

  5. import org.apache.flink.streaming.api.scala._

  6. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011

  7. 

  8. import java.util.Properties

  9. 

  10. /**

  11.  * @Package com.avro.FlinkKafka

  12.  * @File ：UserBehaviorProducerFlink.java

  13.  * @author 大数据老哥

  14.  * @date 2021/1/8 21:38

  15.  * @version V1.0

  16.  */

  17. object UserBehaviorProducerFlink {

  18.   def main(args: Array[String]): Unit = {

  19.     val env = StreamExecutionEnvironment.getExecutionEnvironment

  20.     val value = env.readTextFile("./data/UserBehavior.csv")

  21.     val users: DataStream[UserBehavior] = value.map(row => {

  22.       val arr = row.split(",")

  23.       val behavior = new UserBehavior()

  24.       behavior.setUserId(arr(0).toLong)

  25.       behavior.setItemId(arr(1).toLong)

  26.       behavior.setCategoryId(arr(2).toInt)

  27.       behavior.setBehavior(arr(3))

  28.       behavior.setTimestamp(arr(4).toLong)

  29.       behavior

  30.     })

  31.     val prop = new Properties()

  32.     prop.setProperty("bootstrap.servers", "node01:9092,node02:9092,node03:9092")

  33.     //4.连接Kafka

  34.     val producer: FlinkKafkaProducer011[UserBehavior] = new FlinkKafkaProducer011[UserBehavior]("UserBehaviorKafka", new SimpleAvroSchemaFlink(), prop)

  35.     //5.将数据打入kafka

  36.     users.addSink(producer)

  37.     //6.执行任务

  38.     env.execute("UserBehaviorProducerFlink")

  39.   }

  40. }
复制代码

5.5 启动运行
2021-08-10_174149.jpg


需要源码的请去GitHub 自行下载 https://github.com/lhh2002/Flink_Avro

小结

其实我在实现这个功能的时候也是蒙的，不会难道就不学了吗，肯定不是呀。我在5.2提出的那个问题的时候其实是我自己亲身经历过的。首先遇到了问题不要想着怎么放弃，而是想想怎么解决，当时我的思路看源码看别人写的。最后经过不懈的努力也终成功了，我在这里为大家提供Flink面试题需要的朋友可以去下面GitHub去下载，信自己，努力和汗水总会能得到回报的。我是大数据老哥，我们下期见~~~

作者：大数据老哥
来源：https://blog.csdn.net/qq_43791724/article/details/112371170

最新经典文章，欢迎关注公众号



回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
levycui

超级版主

关注

  • 653

    主题

  • 1166

    帖子

  • 97

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000