一.元组(Tuples and Case Classes )
对java来说Tuples是flink自带的一种类, 对于scala来说flink没有提供类似Tuples的类, 因为scala天生自带了一种特殊类 case class.
主要说说java版的Tuples, Java API 提供从Tuple1最高到Tuple25. 元组的每个字段都可以是任意 Flink 类型, 1 25这个数字的意思是参数的个数.
Tuple1 t1;
Tuple2<String,String> t2;
Tuple3<String,Integer,Long> t3;
访问Tuple中的数据flink提供了便捷的方法,比如:tuple.getField(int position)。字段索引从 0 开始, 或者tuple.f格式, f后面跟数字,也是从0开始.
比如 Tuple3<String, String,Integer> t3=new Tuple3<String,
String,Integer>(“张三”, “男”,20)
如果要访问年龄有下面两种方法:
- t3.f2
- t3.getField(2)
case class WordCount(word: String, count: Int)
val input = env.fromElements(
WordCount("hello", 1),
WordCount("world", 2)) // Case Class Data Set
2.java版
DataStream<Tuple2<String, Integer>> wordCounts = env.fromElements(
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("world", 2));
wordCounts.map(new MapFunction<Tuple2<String, Integer>, Integer>() {
@Override
public Integer map(Tuple2<String, Integer> value) throws Exception {
return value.f1;
}
});
我们来看看java的Tuple,Tuple是一个实现了java序列化接口的一个顶层接口,Tuple2 Tupe3 …Tuple25是实现了Tuple接口的具体的类
Tuple接口实现了java的序列化接口,public abstract class Tuple implements java.io.Serializable
下面是Tuple2的源码:
/*
* 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.
*/
// --------------------------------------------------------------
// THIS IS A GENERATED SOURCE FILE. DO NOT EDIT!
// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
// --------------------------------------------------------------
package org.apache.flink.api.java.tuple;
import org.apache.flink.annotation.Public;
import org.apache.flink.util.StringUtils;
/**
* A tuple with 2 fields. Tuples are strongly typed; each field may be of a separate type. The
* fields of the tuple can be accessed directly as public fields (f0, f1, ...) or via their position
* through the {@link #getField(int)} method. The tuple field positions start at zero.
*
* <p>Tuples are mutable types, meaning that their fields can be re-assigned. This allows functions
* that work with Tuples to reuse objects in order to reduce pressure on the garbage collector.
*
* <p>Warning: If you subclass Tuple2, then be sure to either
*
* <ul>
* <li>not add any new fields, or
* <li>make it a POJO, and always declare the element type of your DataStreams/DataSets to your
* descendant type. (That is, if you have a "class Foo extends Tuple2", then don't use
* instances of Foo in a DataStream<Tuple2> / DataSet<Tuple2>, but declare it as
* DataStream<Foo> / DataSet<Foo>.)
* </ul>
*
* @see Tuple
* @param <T0> The type of field 0
* @param <T1> The type of field 1
*/
@Public
public class Tuple2<T0, T1> extends Tuple {
private static final long serialVersionUID = 1L;
/** Field 0 of the tuple. */
public T0 f0;
/** Field 1 of the tuple. */
public T1 f1;
/** Creates a new tuple where all fields are null. */
public Tuple2() {}
/**
* Creates a new tuple and assigns the given values to the tuple's fields.
*
* @param f0 The value for field 0
* @param f1 The value for field 1
*/
public Tuple2(T0 f0, T1 f1) {
this.f0 = f0;
this.f1 = f1;
}
@Override
public int getArity() {
return 2;
}
@Override
@SuppressWarnings("unchecked")
public <T> T getField(int pos) {
switch (pos) {
case 0:
return (T) this.f0;
case 1:
return (T) this.f1;
default:
throw new IndexOutOfBoundsException(String.valueOf(pos));
}
}
@Override
@SuppressWarnings("unchecked")
public <T> void setField(T value, int pos) {
switch (pos) {
case 0:
this.f0 = (T0) value;
break;
case 1:
this.f1 = (T1) value;
break;
default:
throw new IndexOutOfBoundsException(String.valueOf(pos));
}
}
/**
* Sets new values to all fields of the tuple.
*
* @param f0 The value for field 0
* @param f1 The value for field 1
*/
public void setFields(T0 f0, T1 f1) {
this.f0 = f0;
this.f1 = f1;
}
/**
* Returns a shallow copy of the tuple with swapped values.
*
* @return shallow copy of the tuple with swapped values
*/
public Tuple2<T1, T0> swap() {
return new Tuple2<T1, T0>(f1, f0);
}
// -------------------------------------------------------------------------------------------------
// standard utilities
// -------------------------------------------------------------------------------------------------
/**
* Creates a string representation of the tuple in the form (f0, f1), where the individual
* fields are the value returned by calling {@link Object#toString} on that field.
*
* @return The string representation of the tuple.
*/
@Override
public String toString() {
return "("
+ StringUtils.arrayAwareToString(this.f0)
+ ","
+ StringUtils.arrayAwareToString(this.f1)
+ ")";
}
/**
* Deep equality for tuples by calling equals() on the tuple members.
*
* @param o the object checked for equality
* @return true if this is equal to o.
*/
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (!(o instanceof Tuple2)) {
return false;
}
@SuppressWarnings("rawtypes")
Tuple2 tuple = (Tuple2) o;
if (f0 != null ? !f0.equals(tuple.f0) : tuple.f0 != null) {
return false;
}
if (f1 != null ? !f1.equals(tuple.f1) : tuple.f1 != null) {
return false;
}
return true;
}
@Override
public int hashCode() {
int result = f0 != null ? f0.hashCode() : 0;
result = 31 * result + (f1 != null ? f1.hashCode() : 0);
return result;
}
/**
* Shallow tuple copy.
*
* @return A new Tuple with the same fields as this.
*/
@Override
@SuppressWarnings("unchecked")
public Tuple2<T0, T1> copy() {
return new Tuple2<>(this.f0, this.f1);
}
/**
* Creates a new tuple and assigns the given values to the tuple's fields. This is more
* convenient than using the constructor, because the compiler can infer the generic type
* arguments implicitly. For example: {@code Tuple3.of(n, x, s)} instead of {@code new
* Tuple3<Integer, Double, String>(n, x, s)}
*/
public static <T0, T1> Tuple2<T0, T1> of(T0 f0, T1 f1) {
return new Tuple2<>(f0, f1);
}
}
所以flink针对Tuple的序列化,底层还是用的java的序列化,并没有用其他的序列化框架.
二.java或者scala 遵循下述规范的类(POJOs )
普通类有以下要求:
- 必须是public 类
- 必须有一个不带参数的默认构造函数
- 字段必须也是公共的,或者提供get/set方法
- 字段的类型必须被注册的序列化器支持
下面是例子代码:
public class WordWithCount {
public String word;
public int count;
public WordWithCount() {}
public WordWithCount(String word, int count) {
this.word = word;
this.count = count;
}
}
DataStream<WordWithCount> wordCounts = env.fromElements(
new WordWithCount("hello", 1),
new WordWithCount("world", 2));
wordCounts.keyBy(value -> value.word);
下面是scala
class WordWithCount(var word: String, var count: Int) {
//无参辅助构造器
def this() {
this(null, -1)
}
}
val input = env.fromElements(
//下面这种是直接调用的主构造器,关于scala构造器请参考我的其他文章
new WordWithCount("hello", 1),
new WordWithCount("world", 2)) // Case Class Data Set
input.keyBy(_.word)
下面来说说工作原理:对于你自己定义的普通类,flink首先会对你的这个类做类的检测,比如针对第一条检测是否是public 修饰的类–>Modifier.isPublic([类].getModifiers()), 检测完了之后发现符合上述四条规则,那么就会对当前类调用PojoSerializer 序列化器进行封装,下面是继承关系:
public final class PojoSerializer extends TypeSerializer {…}
public abstract class TypeSerializer implements Serializable{…}
可以看出最后用的序列化还是java的序列化. TypeSerializer是一个顶层接口,基本上所有的序列化的类都是TypeSerializer的一种实现包括PojoSerializer,下面是一些实现了TypeSerializer的类.
如果检测不符合上述四条规则,那么flink默认的序列化器是上图中的:KryoSerializer ,这个序列化器就是用的 Kryo框架.打开KryoSerializer 类发现有下面的注释:
A type serializer that serializes its type using the Kryo serialization framework (https://github.com/EsotericSoftware/kryo).
This serializer is intended as a fallback serializer for the cases that are not covered by the basic types, tuples, and POJOs.
Type parameters:
– The type to be serialized.
public class KryoSerializer extends TypeSerializer {…代码省略}
三.原始类型(Primitive Types )
flink支持所有scala/java 的所有原始类型:Integer String Double
四.通用类(General Class Types)
java/scala 不遵守二中所说的规范,那么scala会将此类按照统一的序列化标准进行序列化,这个序列化标准采用的序列化框架是Kryo
五.flink内置的Values类型
你需要实现org.apache.flink.types.Value 接口的 read 和write方法. 和通用类(General Class Types) 相比Values接口序列化方式更加高效.
Flink提供了预定义的Value类型，与基本数据类型相对应。(ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue)。这些Value类型充当基本数据类型的可变变量:它们的值可以更改，从而允许程序员重用对象，减轻垃圾收集器的压力。
下图展示了预定义的类型和接口Value的关系:
六.Hadoop Writables
您可以使用实现org.apache.hadoop.Writable接口的类型。write()和方法中定义的序列化逻辑readFields()将用于序列化。
七.特殊类型
您可以使用特殊类型，包括 Scala 的Either、Option和Try. Java API 有自己的自定义实现Either。与 Scala 类似Either，它表示两种可能类型的值，左或右。 Either对于需要输出两种不同类型记录的错误处理或运算符非常有用。
关于java泛型类型擦除的问题’
Java 编译器在编译后丢弃了很多泛型类型信息。这在 Java中称为类型擦除。这意味着在运行时，对象的实例不再知道其泛型类型。例如，在 JVM 中DataStream和的实例DataStream看起来相同。
Flink 在准备程序执行时（调用程序的 main 方法时）需要类型信息。Flink Java API 试图以各种方式重构被丢弃的类型信息，并将其显式存储在数据集和运算符中。您可以通过检索类型DataStream.getType()。该方法返回一个 的实例TypeInformation，这是 Flink 内部表示类型的方式。
类型推断有其局限性，在某些情况下需要程序员代码来“配合”。，例如 ExecutionEnvironment.fromCollection(),您可以在其中传递描述类型的参数。
但是像MapFunction<I, O>这样的泛型函数可能需要额外的类型信息。
常见的问题
- 注册子类型：如果函数签名只描述超类型，但在执行过程中实际上使用了超类型的子类型，那么让 Flink
了解这些子类型可能会大大提高性能。为此，请.registerType(clazz)为每个子类型调用StreamExecutionEnvironmentor
ExecutionEnvironment。 - 注册自定义序列化器： Flink 回退到Kryo来处理它自己不能透明处理的类型。并非所有类型都由 Kryo（因此也由
Flink）无缝处理。例如，许多 Google Guava
集合类型在默认情况下无法正常工作。解决方案是为导致问题的类型注册额外的序列化程序。这句话是说有个特殊的类型既不是Integer也不是String,而是特殊的自定义的类型比如MyInt此时我们可以针对MyInt注册一个自定义的序列化器.调用。.getConfig().addDefaultKryoSerializer(clazz,
serializer)-许多库中提供了其他 Kryo
序列化程序。有关使用自定义序列化程序的更多详细信息，请参阅自定义序列化程序。此功能其实不常用,一般来说我们只会自定义对象,很少会自定义类型. - 添加类型提示：有时，尽管 Flink 无法推断出泛型类型，但用户必须传递类型提示。这通常只在 Java API
中是必需的。类型提示部分更详细地描述了这一点。 - 手动创建TypeInformation:这对于某些 API 调用可能是必要的，因为 Java 的泛型类型擦除，Flink
无法推断数据类型。有关详细信息，请参阅创建 TypeInformation 或 TypeSerializer 。
关于TypeInfomation
flink实现了自己的类型系统,方便做类型检查.以及扁平化对象到schem ,TypeInfomation就是所有类型的基类,TypeInfomation描述了当前类型的类型是什么,以及它序列化器用的什么,下图是是实现此接口的基类:
我们来看一个BasicTypeInfo:
BasicTypeInfo for primitive types (int, long, double, byte, …), String, Date, Void, BigInteger, and BigDecimal.(BasicTypeInfo 是java基础类型的封装类,比如 int long double byte)
public static final BasicTypeInfo<Boolean> BOOLEAN_TYPE_INFO =
new BasicTypeInfo<>(
Boolean.class,
new Class<?>[] {},
BooleanSerializer.INSTANCE,
BooleanComparator.class);
注意参数:BooleanSerializer 这个序列化器其实就是实现了TypeSerializer.
上面说过: TypeSerializer是一个顶层接口,基本上所有的序列化的类都是TypeSerializer的一种实现.