Flink - TypeInformation

简介:

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+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
4月前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队,全球领先的流计算引擎缔造者,支撑双11万亿级数据处理,推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才,地点覆盖北京、杭州、上海。技术深度参与开源核心,打造企业级实时计算解决方案,助力全球企业实现毫秒洞察。
514 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。
|
存储 监控 大数据
阿里云实时计算Flink在多行业的应用和实践
本文整理自 Flink Forward Asia 2023 中闭门会的分享。主要分享实时计算在各行业的应用实践,对回归实时计算的重点场景进行介绍以及企业如何使用实时计算技术,并且提供一些在技术架构上的参考建议。
1514 7
阿里云实时计算Flink在多行业的应用和实践
|
存储 分布式计算 流计算
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
本文介绍了阿里云开源大数据团队在实时计算领域的最新成果——向量化流计算引擎Flash。文章主要内容包括:Apache Flink 成为业界流计算标准、Flash 核心技术解读、性能测试数据以及在阿里巴巴集团的落地效果。Flash 是一款完全兼容 Apache Flink 的新一代流计算引擎,通过向量化技术和 C++ 实现,大幅提升了性能和成本效益。
3742 73
实时计算 Flash – 兼容 Flink 的新一代向量化流计算引擎
|
SQL 消息中间件 Kafka
实时计算 Flink版产品使用问题之如何在EMR-Flink的Flink SOL中针对source表单独设置并行度
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
zdl
|
消息中间件 运维 大数据
大数据实时计算产品的对比测评:实时计算Flink版 VS 自建Flink集群
本文介绍了实时计算Flink版与自建Flink集群的对比,涵盖部署成本、性能表现、易用性和企业级能力等方面。实时计算Flink版作为全托管服务,显著降低了运维成本,提供了强大的集成能力和弹性扩展,特别适合中小型团队和业务波动大的场景。文中还提出了改进建议,并探讨了与其他产品的联动可能性。总结指出,实时计算Flink版在简化运维、降低成本和提升易用性方面表现出色,是大数据实时计算的优选方案。
zdl
555 56
|
11月前
|
消息中间件 关系型数据库 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的过程。最后,介绍了未来规划,如脏数据处理、数据限流及扩展数据源支持。
703 0
Flink CDC 在阿里云实时计算Flink版的云上实践
|
12月前
|
存储 关系型数据库 BI
实时计算UniFlow:Flink+Paimon构建流批一体实时湖仓
实时计算架构中,传统湖仓架构在数据流量管控和应用场景支持上表现良好,但在实际运营中常忽略细节,导致新问题。为解决这些问题,提出了流批一体的实时计算湖仓架构——UniFlow。该架构通过统一的流批计算引擎、存储格式(如Paimon)和Flink CDC工具,简化开发流程,降低成本,并确保数据一致性和实时性。UniFlow还引入了Flink Materialized Table,实现了声明式ETL,优化了调度和执行模式,使用户能灵活调整新鲜度与成本。最终,UniFlow不仅提高了开发和运维效率,还提供了更实时的数据支持,满足业务决策需求。
|
人工智能 Apache 流计算
Flink Forward Asia 2024 上海站|探索实时计算新边界
Flink Forward Asia 2024 即将盛大开幕!11 月 29 至 30 日在上海举行,大会聚焦 Apache Flink 技术演进与未来规划,涵盖流式湖仓、流批一体、Data+AI 融合等前沿话题,提供近百场专业演讲。立即报名,共襄盛举!官网:https://asia.flink-forward.org/shanghai-2024/
1334 33
Flink Forward Asia 2024 上海站|探索实时计算新边界