Flink - TypeInformation

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介:

Flink 自己创建一套独立的类型系统,

参考, https://ci.apache.org/projects/flink/flink-docs-release-0.10/internals/types_serialization.html

为何要自己搞一套,而不像其他的平台一样让编程语言或serialization framework来天然做掉?

Flink tries to know as much information about what types enter and leave user functions as possible. This stands in contrast to the approach to just assuming nothing and letting the programming language and serialization framework handle all types dynamically.

  • To allow using POJOs and grouping/joining them by referring to field names, Flink needs the type information to make checks (for typos and type compatibility) before the job is executed.

  • The more we know, the better serialization and data layout schemes the compiler/optimizer can develop. That is quite important for the memory usage paradigm in Flink (work on serialized data inside/outside the heap and make serialization very cheap).

  • For the upcoming logical programs (see roadmap draft) we need this to know the “schema” of functions.

  • Finally, it also spares users having to worry about serialization frameworks and having to register types at those frameworks.

Note. POJOs是什么?Plain Old Java Object(简单的Java对象),即轻量java对象的花式叫法

主要的理由,

第一是要做类型检查,Flink支持比较灵活的基于field的join或group,需要先检查这个field是否可以作为key,或这个field是否可以做join或group

第二是性能优化,便于使用更好的序列化和数据的layout

Flink主要定义如下几种类型,

Internally, Flink makes the following distinctions between types:

  • Basic types: All Java primitives and their boxed form, plus voidString, and Date.

  • Primitive arrays and Object arrays

  • Composite types

    • Flink Java Tuples (part of the Flink Java API)

    • Scala case classes (including Scala tuples)

    • POJOs: classes that follow a certain bean-like pattern

  • Scala auxiliary types (Option, Either, Lists, Maps, …)

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

基本类型

数组(包含Primitive数组和对象数组)

组合类型,包含Flink Tuples, Scala case classes, 和POJOS

Scala增加的辅助类型

泛型,这个Flink不处理,而是用kryo

这里尤其需要注意POJOs,因为它的field是可以直接用name引用的,非常方便

dataSet.join(another).where("name").equalTo("personName")

那么对于Flink的准确的POJO的定义是什么?

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

很简单,只要满足上面的规范,就支持“by-name” field referencing

文档里面还描述了在Scala和Java API中的类型问题,

对于Scala,用manifest或typetag来解决了泛型擦除的问题,所以主要是Flink用macro实现了TypeInformation,便于使用

对于Java,就必须要解决泛型擦除的问题,

DataSet<SomeType> result = dataSet
    .map(new MyGenericNonInferrableFunction<Long, SomeType>())
        .returns(SomeType.class);
比如,上面的日志,如果不加最后的hints,在runtime其实是无法知道SomeType是什么的,在编译的时候已经被erase成Object

所以Flink使用returns原语来增加hints

 

来看看源码,

基类为,

package org.apache.flink.api.common.typeinfo;
TypeInformation

目的, This type information class acts as the tool

to generate serializers and comparators
to perform semantic checks such as whether the fields that are uses as join/grouping keys actually exist.
bridges between the programming languages object model and a logical flat schema

前两个目的好理解,

最后一个目的,搞清两个概念,

arity,the number of fields it contains directly 
total number of fields,number of fields in the entire schema of this type, including nested types

举个例子,

* public class InnerType {
* public int id;
* public String text;
* }
*
* public class OuterType {
* public long timestamp;
* public InnerType nestedType;
* }

对于Inner type,arity和fields都是2

但对于OuterType,虽然arity是2,但fields是3,因为要把嵌套类型的fields也算上,这就是把编程语言对象模型转换为flat的逻辑schema

如何算fields的规则如下:

*   <li>Basic types are indivisible and are considered a single field.</li>
* <li>Arrays and collections are one field</li>
* <li>Tuples and case classes represent as many fields as the class has fields</li>

 

IntegerTypeInfo
用这个作为例子,分析一下
public class IntegerTypeInfo<T> extends NumericTypeInfo<T> 
public abstract class NumericTypeInfo<T> extends BasicTypeInfo<T> 
public class BasicTypeInfo<T> extends TypeInformation<T> implements AtomicType<T>

可以看到Integer最终继承到BasicType,BasicType除了继承TypeInformation还实现AtomicType接口,

public interface AtomicType<T> {   

TypeComparator<T> createComparator(boolean sortOrderAscending, ExecutionConfig executionConfig);
}
* An atomic type is a type that is treated as one indivisible unit and where the entire type acts
* as a key.
* In contrast to atomic types are composite types, where the type information is aware of the individual
* fields and individual fields may be used as a key.
atomic类型就是不可分的类型,不像composite类型还包含其他的field,所以atomic本身整个作为key,基本类型如int肯定是属于atomic类型的
 
在BasicTypeInfo中定义了所有基本类型的TypeInfo,
复制代码
    public static final BasicTypeInfo<String> STRING_TYPE_INFO = new BasicTypeInfo<String>(String.class, new Class<?>[]{}, StringSerializer.INSTANCE, StringComparator.class);
    public static final BasicTypeInfo<Boolean> BOOLEAN_TYPE_INFO = new BasicTypeInfo<Boolean>(Boolean.class, new Class<?>[]{}, BooleanSerializer.INSTANCE, BooleanComparator.class);
    public static final BasicTypeInfo<Byte> BYTE_TYPE_INFO = new IntegerTypeInfo<Byte>(Byte.class, new Class<?>[]{Short.class, Integer.class, Long.class, Float.class, Double.class, Character.class}, ByteSerializer.INSTANCE, ByteComparator.class);
    public static final BasicTypeInfo<Short> SHORT_TYPE_INFO = new IntegerTypeInfo<Short>(Short.class, new Class<?>[]{Integer.class, Long.class, Float.class, Double.class, Character.class}, ShortSerializer.INSTANCE, ShortComparator.class);
    public static final BasicTypeInfo<Integer> INT_TYPE_INFO = new IntegerTypeInfo<Integer>(Integer.class, new Class<?>[]{Long.class, Float.class, Double.class, Character.class}, IntSerializer.INSTANCE, IntComparator.class);
    public static final BasicTypeInfo<Long> LONG_TYPE_INFO = new IntegerTypeInfo<Long>(Long.class, new Class<?>[]{Float.class, Double.class, Character.class}, LongSerializer.INSTANCE, LongComparator.class);
    public static final BasicTypeInfo<Float> FLOAT_TYPE_INFO = new FractionalTypeInfo<Float>(Float.class, new Class<?>[]{Double.class}, FloatSerializer.INSTANCE, FloatComparator.class);
    public static final BasicTypeInfo<Double> DOUBLE_TYPE_INFO = new FractionalTypeInfo<Double>(Double.class, new Class<?>[]{}, DoubleSerializer.INSTANCE, DoubleComparator.class);
    public static final BasicTypeInfo<Character> CHAR_TYPE_INFO = new BasicTypeInfo<Character>(Character.class, new Class<?>[]{}, CharSerializer.INSTANCE, CharComparator.class);
    public static final BasicTypeInfo<Date> DATE_TYPE_INFO = new BasicTypeInfo<Date>(Date.class, new Class<?>[]{}, DateSerializer.INSTANCE, DateComparator.class);
    public static final BasicTypeInfo<Void> VOID_TYPE_INFO = new BasicTypeInfo<Void>(Void.class, new Class<?>[]{}, VoidSerializer.INSTANCE, null);
复制代码

可以看到Byte,short,int,long都用的是IntegerTypeInfo

创建的4个参数分别为,以INT_TYPE_INFO为例,

class对象,Integer.class

可能被cast成的类型,所以对于Integer,被cast成long,float,double,character都是可以的

Serializer对象

Comparator对象

可以看到flink重新封装了所有对象的Serializer和Comparator

我们看下LongSerializer,

@Override
public void serialize(Long record, DataOutputView target) throws IOException {
    target.writeLong(record.longValue());
}

很高效的,对于Long,只会序列化真正的longValue,而不会存多余的东西

 

NumericTypeInfo,只是一种特殊的BasicTypeInfo

复制代码
    private static final Set<Class<?>> numericalTypes = Sets.<Class<?>>newHashSet(
            Integer.class,
            Long.class,
            Double.class,
            Byte.class,
            Short.class,
            Float.class,
            Character.class
    );
复制代码

只有上面这几种class对象,才被认为是NumericTypeInfo

而IntegerTypeInfo,只是范围的进一步缩小,

复制代码
    private static final Set<Class<?>> integerTypes = Sets.<Class<?>>newHashSet(
            Integer.class,
            Long.class,
            Byte.class,
            Short.class,
            Character.class
    );
复制代码
 

除了上面的AtomicType,还有如array的typeinfo

比如,BasicArrayTypeInfo

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
6月前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
4月前
|
存储 分布式计算 流计算
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
本文介绍了阿里云开源大数据团队在实时计算领域的最新成果——向量化流计算引擎Flash。文章主要内容包括:Apache Flink 成为业界流计算标准、Flash 核心技术解读、性能测试数据以及在阿里巴巴集团的落地效果。Flash 是一款完全兼容 Apache Flink 的新一代流计算引擎,通过向量化技术和 C++ 实现,大幅提升了性能和成本效益。
1942 73
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
|
8月前
|
存储 监控 大数据
阿里云实时计算Flink在多行业的应用和实践
本文整理自 Flink Forward Asia 2023 中闭门会的分享。主要分享实时计算在各行业的应用实践,对回归实时计算的重点场景进行介绍以及企业如何使用实时计算技术,并且提供一些在技术架构上的参考建议。
994 7
阿里云实时计算Flink在多行业的应用和实践
|
2月前
|
消息中间件 关系型数据库 MySQL
Flink CDC 在阿里云实时计算Flink版的云上实践
本文整理自阿里云高级开发工程师阮航在Flink Forward Asia 2024的分享,重点介绍了Flink CDC与实时计算Flink的集成、CDC YAML的核心功能及应用场景。主要内容包括:Flink CDC的发展及其在流批数据处理中的作用;CDC YAML支持的同步链路、Transform和Route功能、丰富的监控指标;典型应用场景如整库同步、Binlog原始数据同步、分库分表同步等;并通过两个Demo展示了MySQL整库同步到Paimon和Binlog同步到Kafka的过程。最后,介绍了未来规划,如脏数据处理、数据限流及扩展数据源支持。
232 0
Flink CDC 在阿里云实时计算Flink版的云上实践
zdl
|
4月前
|
消息中间件 运维 大数据
大数据实时计算产品的对比测评:实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比,涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务,显著降低了运维成本,提供了强大的集成能力和弹性扩展,特别适合中小型团队和业务波动大的场景。文中还提出了改进建议,并探讨了与其他产品的联动可能性。总结指出,实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色,是大数据实时计算的优选方案。
zdl
236 56
|
3月前
|
存储 关系型数据库 BI
实时计算UniFlow:Flink+Paimon构建流批一体实时湖仓
实时计算架构中,传统湖仓架构在数据流量管控和应用场景支持上表现良好,但在实际运营中常忽略细节,导致新问题。为解决这些问题,提出了流批一体的实时计算湖仓架构——UniFlow。该架构通过统一的流批计算引擎、存储格式(如Paimon)和Flink CDC工具,简化开发流程,降低成本,并确保数据一致性和实时性。UniFlow还引入了Flink Materialized Table,实现了声明式ETL,优化了调度和执行模式,使用户能灵活调整新鲜度与成本。最终,UniFlow不仅提高了开发和运维效率,还提供了更实时的数据支持,满足业务决策需求。
|
7月前
|
SQL 消息中间件 Kafka
实时计算 Flink版产品使用问题之如何在EMR-Flink的Flink SOL中针对source表单独设置并行度
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
SQL 运维 数据可视化
阿里云实时计算Flink版产品体验测评
阿里云实时计算Flink基于Apache Flink构建,提供一站式实时大数据分析平台,支持端到端亚秒级实时数据分析,适用于实时大屏、实时报表、实时ETL和风控监测等场景,具备高性价比、开发效率、运维管理和企业安全等优势。
|
6月前
|
人工智能 Apache 流计算
Flink Forward Asia 2024 上海站|探索实时计算新边界
Flink Forward Asia 2024 即将盛大开幕!11 月 29 至 30 日在上海举行,大会聚焦 Apache Flink 技术演进与未来规划,涵盖流式湖仓、流批一体、Data+AI 融合等前沿话题,提供近百场专业演讲。立即报名,共襄盛举!官网:https://asia.flink-forward.org/shanghai-2024/
1037 33
Flink Forward Asia 2024 上海站|探索实时计算新边界
|
5月前
|
运维 搜索推荐 数据安全/隐私保护
阿里云实时计算Flink版测评报告
阿里云实时计算Flink版在用户行为分析与标签画像场景中表现出色,通过实时处理电商平台用户行为数据,生成用户兴趣偏好和标签,提升推荐系统效率。该服务具备高稳定性、低延迟、高吞吐量,支持按需计费,显著降低运维成本,提高开发效率。
128 1