【Flink笔记】kafka-connector消费protobuf格式数据

protoc --proto_path=${SKYWALKING_HOME}/apm-protocol/apm-network/src/main/proto/  --java_out=${RESULT_HOME}/src/main/java $(find ${SKYWALKING_HOME}/apm-protocol/apm-network/src/main/proto/  -iname "*.proto")

public class KafkaJvmMetricDeserializer implements KafkaDeserializationSchema<JVMMetricCollection> {

    @Override
    public boolean isEndOfStream(JVMMetricCollection jvmMetricCollection) {
        return false;
    }

    @Override
    public JVMMetricCollection deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {

        return JVMMetricCollection.parseFrom(consumerRecord.value());
    }

    @Override
    public TypeInformation<JVMMetricCollection> getProducedType() {
        return TypeInformation.of(JVMMetricCollection.class);
    }

}

  env.getConfig().registerTypeWithKryoSerializer(JVMMetricCollection.class, ProtobufSerializer.class);

        FlinkKafkaConsumer<JVMMetricCollection> kafkaConsumer = new FlinkKafkaConsumer<>(topic,
                new KafkaJvmMetricDeserializer(), properties);
        DataStream<JVMMetricCollection> stream = env.addSource(kafkaConsumer);

<dependency>
	<groupId>io.grpc</groupId>
	<artifactId>grpc-protobuf</artifactId>
	<version>${grpc.version}</version>
</dependency>

<dependency>
	<groupId>com.google.protobuf</groupId>
	<artifactId>protobuf-java</artifactId>
	<version>${protobuf.version}</version>
	<scope>compile</scope>
</dependency>

<dependency>
	<groupId>com.twitter</groupId>
	<artifactId>chill-protobuf</artifactId>
	<version>0.7.6</version>
	<!-- exclusions for dependency conversion -->
	<exclusions>
		<exclusion>
			<groupId>com.esotericsoftware.kryo</groupId>
			<artifactId>kryo</artifactId>
		</exclusion>
	</exclusions>
</dependency>