成都建设网站百度官网入口

Flink支持非常完善的数据类型,数据类型的描述信息都是由TypeInformation定义,比较常用的TypeInformation有BasicTypeInfo、TupleTypeInfo、CaseClassTypeInfo、PojoTypeInfo等。

一、原生数据类型

1、BasicTypeInfo

//创建Int类型的数据集
val env = StreamExecutionEnvironment.getExecutionEnvironment
val IntStream:DataStream[Int] = env.fromElements(4,51,2,7)
//创建String类型的数据集
val StringStream:DataStream[String] = env.fromElements("hello","flink")

2、BasicArrayTypeInfo

//通过从数组中创建数据集
val ArrayStream:DataStream[Int] = env.fromCollection(Array(8,5,2,31))
//通过List集合创建数据集
val ListStream:DataStream[Int] = env.fromCollection(List(6,23,63,9))

二、Java Tuples类型

//通过new Tuple2创建元组数据集
val TupleStream:DataStream[Tuple2[String,Int]] = env.fromElements(new Tuple2("唐太宗",1),new Tuple2("汉武帝",2))

三、Scala样例类

Flink通过实现CaseClassTypeInfo支持任意的Scala Case Class,包括Scala tuples类型,支持通过字段名称和位置索引获取指标,不支持存储空值

//scala样例类
case class bing(id:Int,name:String)
object TableTeat {def main(args: Array[String]): Unit = {//val senv = EnvironmentSettings.newInstance().inStreamingMode().build()val senv = StreamExecutionEnvironment.getExecutionEnvironment//val tenv = TableEnvironment.create(senv)val input = senv.fromElements(bing(1,"成吉思汗"),bing(2,"松赞干布"))input.print()senv.execute()}
}

四、POJOs类型

POJOs类可以完成复杂数据结构的定义,Flink通过实现PojoTypeInfo来描述任意的POJOs,包括Java类和Scala类

• POJOs类必须是Public修饰且必须独立定义,不能是内部类;
• POJOs类中必须含有默认空构造器;
• POJOs类中所有的Fields必须是Public或者具有Public修饰的getter和setter方法;
• POJOs类中的字段类型必须是Flink支持的;

public class Person{//字段具有public修饰符public String name;public int age;//具有默认空构造器public Person(){}public Person(String name,int age){this.name = name;this.age = age;}
}

class Person(var name:String,var age:Int){def this(){this(null,-1)}
}

定义好后,就可以在Flink环境中使用

val personStream = env.fromElements(new Person("刘病己",14),new Person("刘秀",25))
personStream.keyBy("name")

五、特殊数据类型

val mapStream = env.fromElements(Map("name" -> "朱元璋","age" -> "18"),Map("name"-> "朱棣","age" -> "24"))