<commit_before>---
title: "Data Types & Serialization"
nav-id: types
nav-parent_id: dev
nav-show_overview: true
nav-pos: 50
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-->

Apache Flink handles data types and serialization in a unique way, containing its own type descriptors,
generic type extraction, and type serialization framework. This document describes the concepts and the rationale behind them.

* This will be replaced by the TOC
{:toc}


## Type handling in Flink

Flink tries to infer a lot of information about the data types that are exchanged and stored during the distributed computation.
Think about it like a database that infers the schema of tables. In most cases, Flink infers all necessary information seamlessly
by itself. Having the type information allows Flink to do some cool things:

* Using POJOs types and grouping / joining / aggregating them by referring to field names (like `dataSet.keyBy("username")`).
  The type information allows Flink to check (for typos and type compatibility) early rather than failing later at runtime.

* The more Flink knows about data types, the better the serialization and data layout schemes are.
  That is quite important for the memory usage paradigm in Flink (work on serialized data inside/outside the heap where ever possible
  and make serialization very cheap).

* Finally, it also spares users in the majority of cases from worrying about serialization frameworks and having to register types.

In general, the information about data types is needed during the *pre-flight phase* - that is, when the program's calls on `DataStream`
and `DataSet` are made, and before any call to `execute()`, `print()`, `count()`, or `collect()`.


## Most Frequent Issues

The most frequent issues where users need to interact with Flink's data type handling are:

* **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` or `ExecutionEnvironment` for each subtype.

* **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` or `ExecutionEnvironment`.
  Additional Kryo serializers are available in many libraries. See [Custom Serializers]({{ site.baseurl }}/dev/custom_serializers.html) for more details on working with custom 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.

* **Manually creating a `TypeInformation`:** This may be necessary for some API calls where it is not possible for Flink to infer
  the data types due to Java's generic type erasure. See [Creating a TypeInformation or TypeSerializer](#creating-a-typeinformation-or-typeserializer)
  for details.


## Flink's TypeInformation class

The class {% gh_link /flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java "TypeInformation" %}
is the base class for all type descriptors. It reveals some basic properties of the type and can generate serializers
and, in specializations, comparators for the types.
(*Note that comparators in Flink do much more than defining an order - they are basically the utility to handle keys*)

Internally, Flink makes the following distinctions between types:

* Basic types: All Java primitives and their boxed form, plus `void`, `String`, `Date`, `BigDecimal`, and `BigInteger`.

* Primitive arrays and Object arrays

* Composite types

  * Flink Java Tuples (part of the Flink Java API): max 25 fields, null fields not supported

  * Scala *case classes* (including Scala tuples): max 22 fields, 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

* Auxiliary types (Option, Either, Lists, Maps, ...)

* Generic types: These will not be serialized by Flink itself, but by Kryo.

POJOs are of particular interest, because they support the creation of complex types and the use of field
names in the definition of keys: `dataSet.join(another).where("name").equalTo("personName")`.
They are also transparent to the runtime and can be handled very efficiently by Flink.


#### Rules for POJO types

Flink recognizes a data type as a POJO type (and allows "by-name" field referencing) if the following
conditions are fulfilled:

* The class is public and standalone (no non-static inner class)
* The class has a public no-argument constructor
* All non-static, non-transient fields in the class (and all superclasses) are either public (and non-final)
  or have a public getter- and a setter- method that follows the Java beans
  naming conventions for getters and setters.

Note that when a user-defined data type can't be recognized as a POJO type, it must be processed as GenericType and
serialized with Kryo.


#### Creating a TypeInformation or TypeSerializer<commit_msg>[FLINK-11996][scala][docs] Remove warning about max fields in case class<commit_after>---
title: "Data Types & Serialization"
nav-id: types
nav-parent_id: dev
nav-show_overview: true
nav-pos: 50
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-->

Apache Flink handles data types and serialization in a unique way, containing its own type descriptors,
generic type extraction, and type serialization framework. This document describes the concepts and the rationale behind them.

* This will be replaced by the TOC
{:toc}


## Type handling in Flink

Flink tries to infer a lot of information about the data types that are exchanged and stored during the distributed computation.
Think about it like a database that infers the schema of tables. In most cases, Flink infers all necessary information seamlessly
by itself. Having the type information allows Flink to do some cool things:

* Using POJOs types and grouping / joining / aggregating them by referring to field names (like `dataSet.keyBy("username")`).
  The type information allows Flink to check (for typos and type compatibility) early rather than failing later at runtime.

* The more Flink knows about data types, the better the serialization and data layout schemes are.
  That is quite important for the memory usage paradigm in Flink (work on serialized data inside/outside the heap where ever possible
  and make serialization very cheap).

* Finally, it also spares users in the majority of cases from worrying about serialization frameworks and having to register types.

In general, the information about data types is needed during the *pre-flight phase* - that is, when the program's calls on `DataStream`
and `DataSet` are made, and before any call to `execute()`, `print()`, `count()`, or `collect()`.


## Most Frequent Issues

The most frequent issues where users need to interact with Flink's data type handling are:

* **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` or `ExecutionEnvironment` for each subtype.

* **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` or `ExecutionEnvironment`.
  Additional Kryo serializers are available in many libraries. See [Custom Serializers]({{ site.baseurl }}/dev/custom_serializers.html) for more details on working with custom 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.

* **Manually creating a `TypeInformation`:** This may be necessary for some API calls where it is not possible for Flink to infer
  the data types due to Java's generic type erasure. See [Creating a TypeInformation or TypeSerializer](#creating-a-typeinformation-or-typeserializer)
  for details.


## Flink's TypeInformation class

The class {% gh_link /flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java "TypeInformation" %}
is the base class for all type descriptors. It reveals some basic properties of the type and can generate serializers
and, in specializations, comparators for the types.
(*Note that comparators in Flink do much more than defining an order - they are basically the utility to handle keys*)

Internally, Flink makes the following distinctions between types:

* Basic types: All Java primitives and their boxed form, plus `void`, `String`, `Date`, `BigDecimal`, and `BigInteger`.

* Primitive arrays and Object arrays

* Composite 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

* Auxiliary types (Option, Either, Lists, Maps, ...)

* Generic types: These will not be serialized by Flink itself, but by Kryo.

POJOs are of particular interest, because they support the creation of complex types and the use of field
names in the definition of keys: `dataSet.join(another).where("name").equalTo("personName")`.
They are also transparent to the runtime and can be handled very efficiently by Flink.


#### Rules for POJO types

Flink recognizes a data type as a POJO type (and allows "by-name" field referencing) if the following
conditions are fulfilled:

* The class is public and standalone (no non-static inner class)
* The class has a public no-argument constructor
* All non-static, non-transient fields in the class (and all superclasses) are either public (and non-final)
  or have a public getter- and a setter- method that follows the Java beans
  naming conventions for getters and setters.

Note that when a user-defined data type can't be recognized as a POJO type, it must be processed as GenericType and
serialized with Kryo.


#### Creating a TypeInformation or TypeSerializer
{'fraction_non_alphanumeric': 0.06158218853623875, 'fraction_numerical': 0.001815885046581399, 'mean_word_length': 4.6675615212527966, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4416735', 'n_tokens_mistral': 3247, 'n_tokens_neox': 3103, 'n_words': 1766}