You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2025/01/24 06:53:40 UTC
(flink) 02/02: [FLINK-36905][doc] Update Chinese doc on serialization to reflect the latest changes in Flink 2.0
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 68199affe09a70968c3de58e6fee4b7e5c563deb
Author: Zhanghao Chen <m1...@outlook.com>
AuthorDate: Fri Jan 24 14:50:13 2025 +0800
[FLINK-36905][doc] Update Chinese doc on serialization to reflect the latest changes in Flink 2.0
---
.../serialization/custom_serialization.md | 37 +--
...m_serializers.md => third_party_serializers.md} | 41 +--
.../serialization/types_serialization.md | 283 ++++++++-------------
3 files changed, 139 insertions(+), 222 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md
index f8e4840d28a..4148d6e6bb9 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md
@@ -42,8 +42,6 @@ to specify the state's name, as well as information about the type of the state.
It is also possible to completely bypass this and let Flink use your own custom serializer to serialize managed states,
simply by directly instantiating the `StateDescriptor` with your own `TypeSerializer` implementation:
-{{< tabs "ee215ff6-2e21-4a40-a1b4-7f114560546f" >}}
-{{< tab "Java" >}}
```java
public class CustomTypeSerializer extends TypeSerializer<Tuple2<String, Integer>> {...};
@@ -54,20 +52,6 @@ ListStateDescriptor<Tuple2<String, Integer>> descriptor =
checkpointedState = getRuntimeContext().getListState(descriptor);
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class CustomTypeSerializer extends TypeSerializer[(String, Integer)] {...}
-
-val descriptor = new ListStateDescriptor[(String, Integer)](
- "state-name",
- new CustomTypeSerializer)
-)
-
-checkpointedState = getRuntimeContext.getListState(descriptor)
-```
-{{< /tab >}}
-{{< /tabs >}}
## State serializers and schema evolution
@@ -151,7 +135,7 @@ To wrap up, this section concludes how Flink, or more specifically the state bac
abstractions. The interaction is slightly different depending on the state backend, but this is orthogonal
to the implementation of state serializers and their serializer snapshots.
-#### Off-heap state backends (e.g. `RocksDBStateBackend`)
+#### Off-heap state backends (e.g. `EmbeddedRocksDBStateBackend`)
1. **Register new state with a state serializer that has schema _A_**
- the registered `TypeSerializer` for the state is used to read / write state on every state access.
@@ -172,7 +156,7 @@ to the implementation of state serializers and their serializer snapshots.
of the accessed state is migrated all-together before processing continues.
- If the resolution signals incompatibility, then the state access fails with an exception.
-#### Heap state backends (e.g. `MemoryStateBackend`, `FsStateBackend`)
+#### Heap state backends (e.g. `HashMapStateBackend`)
1. **Register new state with a state serializer that has schema _A_**
- the registered `TypeSerializer` is maintained by the state backend.
@@ -284,6 +268,7 @@ public class MapSerializerSnapshot<K, V> extends CompositeTypeSerializerSnapshot
}
```
+
When implementing a new serializer snapshot as a subclass of `CompositeTypeSerializerSnapshot`,
the following three methods must be implemented:
* `#getCurrentOuterSnapshotVersion()`: This method defines the version of
@@ -311,6 +296,7 @@ has outer snapshot information, then all three methods must be implemented.
Below is an example of how the `CompositeTypeSerializerSnapshot` is used for composite serializer snapshots
that do have outer snapshot information, using Flink's `GenericArraySerializer` as an example:
+
```java
public final class GenericArraySerializerSnapshot<C> extends CompositeTypeSerializerSnapshot<C[], GenericArraySerializer> {
@@ -365,6 +351,7 @@ public final class GenericArraySerializerSnapshot<C> extends CompositeTypeSerial
}
```
+
There are two important things to notice in the above code snippet. First of all, since this
`CompositeTypeSerializerSnapshot` implementation has outer snapshot information that is written as part of the snapshot,
the outer snapshot version, as defined by `getCurrentOuterSnapshotVersion()`, must be upticked whenever the
@@ -449,18 +436,18 @@ migrate from the old abstractions. The steps to do this is as follows:
This section is a guide for a method migration from the serializer snapshots that existed before Flink 1.19.
-Before Flink 1.19, when using a customized serializer to process data, the schema compatibility in the old serializer
-(maybe in Flink library) has to meet the future need.
-Or else TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer<T> newSerializer) of the old serializer has to be modified.
-There are no ways to specify the compatibility with the old serializer in the new serializer, which also makes scheme evolution
+Before Flink 1.19, when using a customized serializer to process data, the schema compatibility in the old serializer
+(maybe in Flink library) has to meet the future need.
+Or else TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer<T> newSerializer) of the old serializer has to be modified.
+There are no ways to specify the compatibility with the old serializer in the new serializer, which also makes scheme evolution
not supported in some scenarios.
-So from Flink 1.19, the direction of resolving schema compatibility has been reversed. The old method
-`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` is now removed and needs to be replaced with
+So from Flink 1.19, the direction of resolving schema compatibility has been reversed. The old method
+`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` is now removed and needs to be replaced with
`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializerSnapshot oldSerializerSnapshot)`.
To make this transition, follow these steps:
-1. Implement the `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializerSnapshot oldSerializerSnapshot)` whose logic
+1. Implement the `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializerSnapshot oldSerializerSnapshot)` whose logic
should be same as the original `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)`.
2. Remove the old method `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)`.
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serializers.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers.md
similarity index 74%
rename from docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serializers.md
rename to docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers.md
index 269575c0d67..7af1e3a15b0 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serializers.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers.md
@@ -29,34 +29,37 @@ under the License.
如果在 Flink 程序中使用了 Flink 类型序列化器无法进行序列化的用户自定义类型，Flink 会回退到通用的 Kryo 序列化器。
可以使用 Kryo 注册自己的序列化器或序列化系统，比如 Google Protobuf 或 Apache Thrift。
-使用方法是在 Flink 程序中的 `ExecutionConfig` 注册类类型以及序列化器。
+使用方法是在 Flink 程序中使用配置 [pipeline.serialization-config]({{< ref "docs/deployment/config#pipeline-serialization-config" >}})
+注册类类型以及序列化器：
-```java
-final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-// 为类型注册序列化器类
-env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, MyCustomSerializer.class);
-
-// 为类型注册序列化器实例
-MySerializer mySerializer = new MySerializer();
-env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, mySerializer);
+```yaml
+pipeline.serialization-config:
+ - org.example.MyCustomType: {type: kryo, kryo-type: registered, class: org.example.MyCustomSerializer}
```
-需要确保你的自定义序列化器继承了 Kryo 的序列化器类。
-对于 Google Protobuf 或 Apache Thrift，这一点已经为你做好了：
+你也可以使用代码设置：
```java
+Configuration config = new Configuration();
-final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+// register the class of the serializer as serializer for a type
+config.set(PipelineOptions.SERIALIZATION_CONFIG,
+ List.of("org.example.MyCustomType: {type: kryo, kryo-type: registered, class: org.example.MyCustomSerializer}"));
-// 使用 Kryo 注册 Google Protobuf 序列化器
-env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, ProtobufSerializer.class);
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+```
-// 注册 Apache Thrift 序列化器为标准序列化器
-// TBaseSerializer 需要初始化为默认的 kryo 序列化器
-env.getConfig().addDefaultKryoSerializer(MyCustomType.class, TBaseSerializer.class);
+需要确保你的自定义序列化器继承了 Kryo 的序列化器类。
+对于 Google Protobuf 或 Apache Thrift，这一点已经为你做好了：
-```
+```yaml
+pipeline.serialization-config:
+# register the Google Protobuf serializer with Kryo
+ - org.example.MyCustomProtobufType: {type: kryo, kryo-type: registered, class: com.twitter.chill.protobuf.ProtobufSerializer}
+# register the serializer included with Apache Thrift as the standard serializer
+# TBaseSerializer states it should be initialized as a default Kryo serializer
+ - org.example.MyCustomThriftType: {type: kryo, kryo-type: default, class: com.twitter.chill.thrift.TBaseSerializer}
+````
为了使上面的例子正常工作，需要在 Maven 项目文件中（pom.xml）包含必要的依赖。
为 Apache Thrift 添加以下依赖：
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
index 70679c5888d..23b1e8ec24c 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/types_serialization.md
@@ -36,20 +36,18 @@ Flink places some restrictions on the type of elements that can be in a DataStre
The reason for this is that the system analyzes the types to determine
efficient execution strategies.
-There are seven different categories of data types:
+There are eight different categories of data types:
-1. **Java Tuples** and **Scala Case Classes**
+1. **Java Tuples**
2. **Java POJOs**
3. **Primitive Types**
-4. **Regular Classes**
-5. **Values**
-6. **Hadoop Writables**
-7. **Special Types**
+4. **Common Collection Types**
+5. **Regular Classes**
+6. **Values**
+7. **Hadoop Writables**
+8. **Special Types**
-#### Tuples and Case Classes
-
-{{< tabs "e24bb87b-46e8-4f17-8054-c3400aaa6ddc" >}}
-{{< tab "Java" >}}
+#### Tuples
Tuples are composite types that contain a fixed number of fields with various types.
The Java API provides classes from `Tuple1` up to `Tuple25`. Every field of a tuple
@@ -75,30 +73,9 @@ wordCounts.keyBy(value -> value.f0);
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-
-Scala case classes (and Scala tuples which are a special case of case classes), are composite types that contain a fixed number of fields with various types. Tuple fields are addressed by their 1-offset names such as `_1` for the first field. Case class fields are accessed by their name.
-
-```scala
-case class WordCount(word: String, count: Int)
-val input = env.fromElements(
- WordCount("hello", 1),
- WordCount("world", 2)) // Case Class Data Set
-
-input.keyBy(_.word)
-
-val input2 = env.fromElements(("hello", 1), ("world", 2)) // Tuple2 Data Set
-
-input2.keyBy(value => (value._1, value._2))
-```
-
-{{< /tab >}}
-{{< /tabs >}}
-
#### POJOs
-Java and Scala classes are treated by Flink as a special POJO data type if they fulfill the following requirements:
+Java classes are treated by Flink as a special POJO data type if they fulfill the following requirements:
- The class must be public.
@@ -120,8 +97,6 @@ If you additionally want to ensure that no field of the POJO will be serialized
The following example shows a simple POJO with two public fields.
-{{< tabs "0589f3b3-76d8-4913-9595-276da92cbc77" >}}
-{{< tab "Java" >}}
```java
public class WordWithCount {
@@ -143,32 +118,29 @@ DataStream<WordWithCount> wordCounts = env.fromElements(
wordCounts.keyBy(value -> value.word);
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class WordWithCount(var word: String, var count: Int) {
- def this() {
- this(null, -1)
- }
-}
-val input = env.fromElements(
- new WordWithCount("hello", 1),
- new WordWithCount("world", 2)) // Case Class Data Set
+#### Primitive Types
-input.keyBy(_.word)
+Flink supports all Java primitive types such as `Integer`, `String`, and `Double`.
-```
-{{< /tab >}}
-{{< /tabs >}}
+#### Common Collection Types
-#### Primitive Types
+Flink comes with dedicated serialization support for common Java collection types, which is more efficient than going
+through a general purpose serialization framework by avoiding analysis and serialization of the type metadata.
+Currently, only `Map`, `List`, `Set` and its super interface `Collection`
+are supported. To utilize it, you need to declare the collection type with:
+
+1. Concrete type arguments: e.g. `List<String>` but not `List`, `List<T>`, or `List<?>`, as Flink needs them to dispatch
+ serialization of the element types.
+2. Interface types: e.g. `List<String>` but not `LinkedList<String>`, as Flink does not preserve the underlying
+ implementation types across serialization.
-Flink supports all Java and Scala primitive types such as `Integer`, `String`, and `Double`.
+Other nonqualified collection types will be handled by Flink as general class types. If the implementation types are
+also required to be preserved, you also need to register it with a custom serializer.
#### General Class Types
-Flink supports most Java and Scala classes (API and custom).
+Flink supports most Java classes (API and custom).
Restrictions apply to classes containing fields that cannot be serialized, like file pointers, I/O streams, or other native
resources. Classes that follow the Java Beans conventions work well in general.
@@ -201,7 +173,6 @@ defined in the `write()`and `readFields()` methods will be used for serializatio
#### Special Types
-You can use special types, including Scala's `Either`, `Option`, and `Try`.
The Java API has its own custom implementation of `Either`.
Similarly to Scala's `Either`, it represents a value of two possible types, *Left* or *Right*.
`Either` can be useful for error handling or operators that need to output two different types of records.
@@ -256,13 +227,39 @@ The most frequent issues where users need to interact with Flink's data type han
* **Registering subtypes:** If the function signatures describe only the supertypes, but they actually use subtypes of those during execution,
it may increase performance a lot to make Flink aware of these subtypes.
- For that, call `.registerType(clazz)` on the `StreamExecutionEnvironment` for each subtype.
+ For that, register serializers for each subtype via
+ [pipeline.serialization-config]({{< ref "docs/deployment/config#pipeline-serialization-config" >}}):
+
+```yaml
+pipeline.serialization-config:
+# register serializer for POJO types
+ - org.example.MyCustomType1: {type: pojo, class: org.example.MyCustomSerializer1}
+# register serializer for generic types with Kryo
+ - org.example.MyCustomType2: {type: kryo, kryo-type: registered, class: org.example.MyCustomSerializer2}
+```
+
+You could also programmatically set it as follows (note that the more compact flow-style YAML is used in the code example,
+while block-style YAML list is used in the above example for better readability):
+
+```java
+Configuration config = new Configuration();
+config.set(PipelineOptions.SERIALIZATION_CONFIG,
+ List.of("org.example.MyCustomType1: {type: pojo, class: org.example.MyCustomSerializer1}",
+ "org.example.MyCustomType2: {type: kryo, kryo-type: registered, class: org.example.MyCustomSerializer2}"));
+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+```
* **Registering custom serializers:** Flink falls back to [Kryo](https://github.com/EsotericSoftware/kryo) for the types that it does not handle transparently
by itself. Not all types are seamlessly handled by Kryo (and thus by Flink). For example, many Google Guava collection types do not work well
- by default. The solution is to register additional serializers for the types that cause problems.
- Call `.getConfig().addDefaultKryoSerializer(clazz, serializer)` on the `StreamExecutionEnvironment`.
- Additional Kryo serializers are available in many libraries. See [Custom Serializers]({{< ref "docs/dev/datastream/fault-tolerance/serialization/custom_serializers" >}}) for more details on working with custom serializers.
+ by default. The solution is to register additional serializers for the types that cause problems via
+ [pipeline.serialization-config]({{< ref "docs/deployment/config#pipeline-serialization-config" >}}):
+
+```yaml
+pipeline.serialization-config:
+ - org.example.MyCustomType: {type: kryo, kryo-type: default, class: org.example.MyCustomSerializer}
+```
+
+ Additional Kryo serializers are available in many libraries. See [3rd party serializer]({{< ref "docs/dev/datastream/fault-tolerance/serialization/third_party_serializers" >}}) for more details on working with external serializers.
* **Adding Type Hints:** Sometimes, when Flink cannot infer the generic types despite all tricks, a user must pass a *type hint*. That is generally
only necessary in the Java API. The [Type Hints Section](#type-hints-in-the-java-api) describes that in more detail.
@@ -289,8 +286,6 @@ Internally, Flink makes the following distinctions between types:
* Flink Java Tuples (part of the Flink Java API): max 25 fields, null fields not supported
- * Scala *case classes* (including Scala tuples): null fields not supported
-
* Row: tuples with arbitrary number of fields and support for null fields
* POJOs: classes that follow a certain bean-like pattern
@@ -320,12 +315,8 @@ serialized with Kryo.
#### Creating a TypeInformation or TypeSerializer
-To create a TypeInformation object for a type, use the language specific way:
-
-{{< tabs "013807cc-9f3f-4f91-a770-26b1e5e8c85c" >}}
-{{< tab "Java" >}}
Because Java generally erases generic type information, you need to pass the type to the TypeInformation
-construction:
+construction.
For non-generic types, you can pass the Class:
```java
@@ -338,100 +329,18 @@ TypeInformation<Tuple2<String, Double>> info = TypeInformation.of(new TypeHint<T
```
Internally, this creates an anonymous subclass of the TypeHint that captures the generic information to preserve it
until runtime.
-{{< /tab >}}
-{{< tab "Scala" >}}
-In Scala, Flink uses *macros* that runs at compile time and captures all generic type information while it is
-still available.
-```scala
-// important: this import is needed to access the 'createTypeInformation' macro function
-import org.apache.flink.streaming.api.scala._
-val stringInfo: TypeInformation[String] = createTypeInformation[String]
-
-val tupleInfo: TypeInformation[(String, Double)] = createTypeInformation[(String, Double)]
-```
-
-You can still use the same method as in Java as a fallback.
-{{< /tab >}}
-{{< /tabs >}}
-
-There are two ways to create a TypeSerializer.
+There are two ways to create a TypeSerializer.
The first is to simply call `typeInfo.createSerializer(config)` on the `TypeInformation` object.
-The `config` parameter is of type `ExecutionConfig` and holds the information about the program's registered
-custom serializers. Where ever possibly, try to pass the programs proper ExecutionConfig. You can usually
-obtain it from `DataStream` via calling `getExecutionConfig()`.
+The `config` parameter is of type `SerializerConfig` and holds the information about the program's registered
+custom serializers. Where ever possibly, try to pass the programs proper `SerializerConfig`.
-The second is to use getRuntimeContext().createSerializer(typeInfo) within a function. Inside functions
-(like `MapFunction`), you can get it by making the function a [Rich Function]() and calling
+The second is to use `getRuntimeContext().createSerializer(typeInfo)` within a function. Inside functions
+(like `MapFunction`), you can get it by making the function a
+[Rich Function]({{< ref "docs/dev/datastream/user_defined_functions#rich-functions" >}}) and calling
`getRuntimeContext().createSerializer(typeInfo)`.
---------
---------
-
-## Type Information in the Scala API
-
-Scala has very elaborate concepts for runtime type information though *type manifests* and *class tags*. In
-general, types and methods have access to the types of their generic parameters - thus, Scala programs do
-not suffer from type erasure as Java programs do.
-
-In addition, Scala allows to run custom code in the Scala Compiler through Scala Macros - that means that some Flink
-code gets executed whenever you compile a Scala program written against Flink's Scala API.
-
-We use the Macros to look at the parameter types and return types of all user functions during compilation - that
-is the point in time when certainly all type information is perfectly available. Within the macro, we create
-a *TypeInformation* for the function's return types (or parameter types) and make it part of the operation.
-
-
-#### No Implicit Value for Evidence Parameter Error
-
-In the case where TypeInformation could not be created, programs fail to compile with an error
-stating *"could not find implicit value for evidence parameter of type TypeInformation"*.
-
-A frequent reason if that the code that generates the TypeInformation has not been imported.
-Make sure to import the entire flink.api.scala package.
-```scala
-import org.apache.flink.api.scala._
-```
-
-Another common cause are generic methods, which can be fixed as described in the following section.
-
-
-#### Generic Methods
-
-Consider the following case below:
-
-```scala
-def selectFirst[T](input: DataStream[(T, _)]) : DataStream[T] = {
- input.map { v => v._1 }
-}
-
-val data : DataStream[(String, Long) = ...
-
-val result = selectFirst(data)
-```
-
-For such generic methods, the data types of the function parameters and return type may not be the same
-for every call and are not known at the site where the method is defined. The code above will result
-in an error that not enough implicit evidence is available.
-
-In such cases, the type information has to be generated at the invocation site and passed to the
-method. Scala offers *implicit parameters* for that.
-
-The following code tells Scala to bring a type information for *T* into the function. The type
-information will then be generated at the sites where the method is invoked, rather than where the
-method is defined.
-
-```scala
-def selectFirst[T : TypeInformation](input: DataStream[(T, _)]) : DataStream[T] = {
- input.map { v => v._1 }
-}
-```
-
-
---------
---------
-
## Type Information in the Java API
@@ -477,8 +386,8 @@ Type extraction for Java 8 lambdas works differently than for non-lambdas, becau
with an implementing class that extends the function interface.
Currently, Flink tries to figure out which method implements the lambda and uses Java's generic signatures to
-determine the parameter types and the return type. However, these signatures are not generated for lambdas
-by all compilers (as of writing this document only reliably by the Eclipse JDT compiler from 4.5 onwards).
+determine the parameter types and the return type. However, these signatures are not generated for lambdas by all compilers.
+If you observe unexpected behavior, manually specify the return type using the `returns` method.
#### Serialization of POJO types
@@ -488,54 +397,61 @@ int, long, String etc. are handled by serializers we ship with Flink.
For all other types, we fall back to [Kryo](https://github.com/EsotericSoftware/kryo).
If Kryo is not able to handle the type, you can ask the `PojoTypeInfo` to serialize the POJO using [Avro](https://avro.apache.org).
-To do so, you have to call
+To do so, make sure to include the `flink-avro` module, and set:
-```java
-final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-env.getConfig().enableForceAvro();
+```yaml
+pipeline.force-avro: true
```
Note that Flink is automatically serializing POJOs generated by Avro with the Avro serializer.
-If you want your **entire** POJO Type to be treated by the Kryo serializer, set
+If you want your **entire** POJO Type to be treated by the Kryo serializer, set:
-```java
-final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-env.getConfig().enableForceKryo();
+```yaml
+pipeline.force-kryo: true
```
-If Kryo is not able to serialize your POJO, you can add a custom serializer to Kryo, using
-```java
-env.getConfig().addDefaultKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass);
-```
-
-There are different variants of these methods available.
+If Kryo is not able to serialize your POJO, you can add a custom serializer to Kryo,
+using [pipeline.serialization-config]({{< ref "docs/deployment/config#pipeline-serialization-config" >}}):
+```yaml
+pipeline.serialization-config:
+ - org.example.MyCustomType: {type: kryo, kryo-type: registered, class: org.example.MyCustomSerializer}
+```
## Disabling Kryo Fallback
There are cases when programs may want to explicitly avoid using Kryo as a fallback for generic types. The most
common one is wanting to ensure that all types are efficiently serialized either through Flink's own serializers,
-or via user-defined custom serializers.
+or via user-defined custom serializers. To do that, set:
-The setting below will raise an exception whenever a data type is encountered that would go through Kryo:
-```java
-env.getConfig().disableGenericTypes();
+```yaml
+pipeline.generic-types: false
```
+An exception will be raised whenever a data type is encountered that would go through Kryo.
## Defining Type Information using a Factory
A type information factory allows for plugging-in user-defined type information into the Flink type system.
You have to implement `org.apache.flink.api.common.typeinfo.TypeInfoFactory` to return your custom type information.
-The factory is called during the type extraction phase if the corresponding type has been annotated
-with the `@org.apache.flink.api.common.typeinfo.TypeInfo` annotation.
+The factory is called during the type extraction phase to supply custom type information for the corresponding type.
+
+In a hierarchy of types, the closest factory will be chosen while traversing upwards.
+However, a built-in factory has the highest precedence. A factory also has higher precedence than
+Flink's built-in types, therefore you should know what you are doing.
-Type information factories can be used in both the Java and Scala API.
+You can associate your custom type information factory with the corresponding type via the configuration option
+[pipeline.serialization-config]({{< ref "docs/deployment/config#pipeline-serialization-config" >}}):
-In a hierarchy of types the closest factory
-will be chosen while traversing upwards, however, a built-in factory has highest precedence. A factory has
-also higher precedence than Flink's built-in types, therefore you should know what you are doing.
+```yaml
+pipeline.serialization-config:
+ - org.example.MyCustomType: {type: typeinfo, class: org.example.MyCustomTypeInfoFactory}
+```
+
+Alternatively, you can annotate either the corresponding type or a POJO's field using this type with the
+`@org.apache.flink.api.common.typeinfo.TypeInfo` annotation to have the factory associated.
+Note that the type information factory associated via configuration option will have higher precedence.
The following example shows how to annotate a custom type `MyTuple` and supply custom type information for it using a factory in Java.
@@ -559,6 +475,17 @@ public class MyTupleTypeInfoFactory extends TypeInfoFactory<MyTuple> {
}
```
+Instead of annotating the type itself, which may not be possible for third-party code, you can also
+annotate the usage of this type inside a valid Flink POJO like this:
+```java
+public class MyPojo {
+ public int id;
+
+ @TypeInfo(MyTupleTypeInfoFactory.class)
+ public MyTuple<Integer, String> tuple;
+}
+```
+
The method `createTypeInfo(Type, Map<String, TypeInformation<?>>)` creates type information for the type the factory is targeted for.
The parameters provide additional information about the type itself as well as the type's generic type parameters if available.