-
[Kryo] Kyro Serialization 알아보기Flink 2024. 2. 5. 07:39728x90반응형
- 목차
들어가며.
Kryo 는 Java Serialization Library 입니다.
Json, csv 와 같은 Text 기반의 Serialization 방식이 아닌 Binary Serialization 형식을 취하구요.
Byte 데이터를 활용합니다.
그래서 File 저장이나 Network 통신의 레벨에서 주로 사용됩니다.
간단히 Kyro Serialization 실행해보기.
간단히 Kryo Serialization Framework 를 통해서 직렬화/역직렬화하는 방법에 대해서 알아보도록 하겠습니다.
저는 4.0.1 버전의 모듈을 사용할 예정입니다.
// https://mvnrepository.com/artifact/com.esotericsoftware/kryo implementation group: 'com.esotericsoftware', name: 'kryo', version: '4.0.1'
아래의 예시는 간단한 Kryo Serialization 실습 프로그램입니다.
코드의 내용은 User class 라는 Pojo 를 직렬화/역직렬화하는 코드이구요.
Java IO 의 ByteArrayInputStream, ByteArrayOutputStream 을 사용하여 메모리로 User Class 의 Instance 를 기록합니다.
Kryo Serialization 을 사용하기 위해서는 Kryo 메인 객체를 생성하고, Serializable 한 Class 들을 등록합니다.
그리고 Java IO 의 Stream 들을 통해서 메모리, File, Network 를 통한 직렬화를 수행할 수 있습니다.
package com.westlife.serialization; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; public class TestKryo { public static void main (String[] args) throws IOException { Kryo kryo = new Kryo(); kryo.register(User.class); User user = new User("Andy", 35); Output output = new Output(new ByteArrayOutputStream()); kryo.writeObject(output, user); Input input = new Input(new ByteArrayInputStream(output.toBytes())); User deserializedUser = kryo.readObject(input, User.class); input.close(); output.close(); System.out.println(deserializedUser); } } class User { public User () {} public User (String name, Integer age) { this.name = name; this.age = age; } private String name; private Integer age; public String getName() {return this.name;} public Integer getAge() {return this.age;} public void setName(String name) {this.name = name;} public void setAge(Integer age) {this.age = age;} public String toString () {return String.format("%s %s %s", this.hashCode(), this.name, this.age);} }
< 출력 내용 >
1489069835 Andy 35
FieldSerializer.
Kryo 는 직렬화와 역직렬화를 위해서 Built-In Serializer 와 Custom Serializer 두가지 방식을 사용합니다.
퍼포먼스를 향상시키기 위해서 Custom Serializer 를 사용할 수 있구요.
일반적으로 Built-In Serializer 인 FieldSerializer 가 사용됩니다.
FieldSerializer 의 Full Name 은 com.esotericsoftware.kryo.serializers.FieldSerializer 이구요.
FieldSerializer 는 Kryo 의 기본 Serializer 로 사용됩니다.
Java Reflection 을 통해 Class 의 Field 들을 획득하고 이를 통해서 Serialization 을 수행합니다.
따라서 직렬화/역직렬화를 수행하는 데이터의 Field Type 이 일치하는 것이 중요합니다.
아래 두 클래스는 서로 Field 의 Schema 가 동일하므로 Compatible 합니다.
class User { String name; Integer age; } class Car { String model; Integer purchasedYear; }
반면, 아래 두 타입은 서로 Compatible 하지 않죠.
class User { String name; Integer age; } class City { String name; Long population; Integer location; }
FieldSerializer 로 직렬화/역직렬화되지 않는 두 타입의 데이터를 관련짓기 위해서 Custom Serializer 가 필요하게 됩니다.
Kryo 의 역직렬화는 Class 에 구애받지 않는다.
Kryo 를 통해서 직렬화된 Bytes 는 Schema 가 동일하다면 어떤 타입의 데이터로 역직렬화될 수 있습니다.
예를 들어, 아래처럼 User.class 와 Person.class 가 존재한다고 가정하겠습니다.
그리고 이 둘의 Field 와 Method 는 서로 동일한 Pojo 입니다.
class User { public User () {} public User (String name, Integer age) { this.name = name; this.age = age; } private String name; private Integer age; public String getName() {return this.name;} public Integer getAge() {return this.age;} public void setName(String name) {this.name = name;} public void setAge(Integer age) {this.age = age;} public String toString () {return String.format("%s %s %s", this.hashCode(), this.name, this.age);} } class Person { public Person () {} public Person (String name, Integer age) { this.name = name; this.age = age; } private String name; private Integer age; public String getName() {return this.name;} public Integer getAge() {return this.age;} public void setName(String name) {this.name = name;} public void setAge(Integer age) {this.age = age;} public String toString () {return String.format("%s %s %s", this.hashCode(), this.name, this.age);} }
JVM 레벨에서 User.class 와 Person.class 는 서로 다른 타입의 데이터이지만, Bytes 레벨에서 서로의 값은 동일합니다.
결과적으로 Kryo 의 역직렬화 관점에서 두 타입은 같은 타입으로 여겨집니다.
public class TestKryo { public static void main (String[] args) { Kryo kryo = new Kryo(); kryo.register(User.class); User user = new User("Andy", 35); Output output = new Output(new ByteArrayOutputStream()); kryo.writeObject(output, user); Input userInput = new Input(new ByteArrayInputStream(output.toBytes())); User deserializedUser = kryo.readObject(userInput, User.class); userInput.close(); Input personInput = new Input(new ByteArrayInputStream(output.toBytes())); Person deserializedPerson = kryo.readObject(personInput, Person.class); personInput.close(); output.close(); System.out.println(deserializedUser); System.out.println(deserializedPerson); } }
1354011814 Andy 35 951880373 Andy 35
반응형'Flink' 카테고리의 다른 글
[Flink] Tumbling Time Window 알아보기 (TumblingEventTimeWindows) (0) 2024.02.12 [Flink] Window AllowedLateness 알아보기 (Watermark) (0) 2024.02.12 [Flink] FileSource 알아보기 (json, csv, parquet, avro) (0) 2024.02.04 [Flink] KafkaSource Connector 알아보기 (0) 2024.02.04 [Flink] Stateless Transform Operator 알아보기 (Map, Filter, FlatMap) (0) 2024.01.25