锐客网

  1. 首页 > it技术 > >

Fllink实时计算运用（七）Flink 自定义序列化Protobuf接入实现方案

1. 自定义序列化接入方案（Protobuf） 在实际应用场景中， 会存在各种复杂传输对象，同时要求较高的传输处理性能， 这就需要采用自定义的序列化方式做相应实现， 这里以Protobuf为例做讲解。
功能： kafka对同一Topic的生产与消费，采用Protobuf做序列化与反序列化传输， 验证能否正常解析数据。

  1. 通过protobuf脚本生成JAVA文件
    syntax = "proto3"; option java_package = "com.itcast.flink.connectors.kafka.proto"; option java_outer_classname = "AccessLogProto"; // 消息结构定义 message AccessLog {string ip = 1; string time = 2; string type = 3; string api = 4; string num = 5; }

通过批处理脚本，生成JAVA文件：
@echo off for %%i in (proto/*.proto) do ( d:/TestCode/protoc.exe --proto_path=./proto--java_out=../java./proto/%%i echo generate %%i to java file successfully! )

注意， 路径要配置正确。
  1. 自定义序列化实现
    添加POM依赖：
    org.apache.flink flink-connector-kafka_2.11 1.11.2 com.google.protobuf protobuf-java 3.8.0 org.springframework spring-beans 5.1.8.RELEASE

AccessLog对象：
@Data public class AccessLog implements Serializable {private String ip; private String time; private String type; private String api; private Integer num; }

CustomSerialSchema:
/** * 自定义序列化实现（Protobuf） */ public class CustomSerialSchema implements DeserializationSchema, SerializationSchema {private static final long serialVersionUID = 1L; private transient Charset charset; public CustomSerialSchema() { this(StandardCharsets.UTF_8); }public CustomSerialSchema(Charset charset) { this.charset = checkNotNull(charset); }public Charset getCharset() { return charset; }/** * 反序列化实现 * @param message * @return */ @Override public AccessLog deserialize(byte[] message) { AccessLog accessLog = null; try { AccessLogProto.AccessLog accessLogProto = AccessLogProto.AccessLog.parseFrom(message); accessLog = new AccessLog(); BeanUtils.copyProperties(accessLogProto, accessLog); return accessLog; } catch (Exception e) { e.printStackTrace(); } return accessLog; }@Override public boolean isEndOfStream(AccessLog nextElement) { return false; }/** * 序列化处理 * @param element * @return */ @Override public byte[] serialize(AccessLog element) { AccessLogProto.AccessLog.Builder builder = AccessLogProto.AccessLog.newBuilder(); BeanUtils.copyProperties(element, builder); return builder.build().toByteArray(); }/** * 定义消息类型 * @return */ @Override public TypeInformation getProducedType() { return TypeInformation.of(AccessLog.class); } }

  1. 通过flink对kafka消息生产者的实现
    public class KafkaSinkApplication {public static void main(String[] args) throws Exception {// 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取Socket数据源 DataStreamSource socketStr = env.socketTextStream("localhost", 9911, "\n"); // 3. 转换处理流数据 SingleOutputStreamOperator outputStream = socketStr.map(new MapFunction() { @Override public AccessLog map(String value) throws Exception { System.out.println(value); // 根据分隔符解析数据 String[] arrValue = https://www.it610.com/article/value.split("\t"); // 将数据组装为对象 AccessLog log = new AccessLog(); log.setNum(1); for(int i=0; i

开启Kafka消费者命令行终端，验证生产者的可用性：
[root@flink1 kafka_2.12-1.1.1]# bin/kafka-console-consumer.sh --bootstrap-server10.10.20.132:9092--topic flink-serial 1601649380422GET" getAccount 1601649381422POSTaddOrder 1601649382422POST"

  1. 【Fllink实时计算运用（七）Flink 自定义序列化Protobuf接入实现方案】通过flink对kafka消息订阅者的实现
    public class KafkaSourceApplication {public static void main(String[] args) throws Exception {// 1. 创建运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 设置kafka服务连接信息 Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "10.10.20.132:9092"); properties.setProperty("group.id", "fink_group"); // 3. 创建Kafka消费端 FlinkKafkaConsumer kafkaProducer = new FlinkKafkaConsumer( "flink-serial",// 目标 topic new CustomSerialSchema(),// 自定义序列化 properties); // 4. 读取Kafka数据源 DataStreamSource socketStr = env.addSource(kafkaProducer); socketStr.print().setParallelism(1); // 5. 执行任务 env.execute("job"); }}

通过flink的kafka生产者消息的发送， 对消费者的功能做测试验证。
本文由mirson创作分享，如需进一步交流，请加QQ群：19310171或访问www.softart.cn

    推荐阅读


    上一篇：没有一个春天不会到来

    下一篇：墨，不归