快捷導(dǎo)航

Flink支持哪些數(shù)據(jù)類(lèi)型?

更新時(shí)間：2021年06月09日 15:23:53 作者：藍(lán)天elasticsearch

Apache Flink 以其獨(dú)特的方式來(lái)處理數(shù)據(jù)類(lèi)型以及序列化,這種方式包括它自身的類(lèi)型描述符、泛型類(lèi)型提取以及類(lèi)型序列化框架.本文檔描述了它們背后的概念和基本原理,需要的朋友可以參考下

一、支持的數(shù)據(jù)類(lèi)型

Flink 對(duì)可以在 DataSet 或 DataStream 中的元素類(lèi)型進(jìn)行了一些限制。這樣做的原因是系統(tǒng)會(huì)分析類(lèi)型以確定有效的執(zhí)行策略。

1.Java Tuple 和 Scala Case類(lèi)；

2.Java POJO；

3.基本類(lèi)型；

4.通用類(lèi)；

5.值；

6.Hadoop Writables;

7.特殊類(lèi)型

二、Flink之Tuple類(lèi)型

Tuple類(lèi)型 Tuple 是flink 一個(gè)很特殊的類(lèi)型（元組類(lèi)型），是一個(gè)抽象類(lèi)，共26個(gè)Tuple子類(lèi)繼承Tuple 他們是 Tuple0一直到Tuple25

package org.apache.flink.api.java.tuple;

import java.io.Serializable;
import org.apache.flink.annotation.Public;
import org.apache.flink.types.NullFieldException;

@Public
public abstract class Tuple implements Serializable {
    private static final long serialVersionUID = 1L;
    public static final int MAX_ARITY = 25;
    private static final Class<?>[] CLASSES = new Class[]{Tuple0.class, Tuple1.class, Tuple2.class, Tuple3.class, Tuple4.class, Tuple5.class, Tuple6.class, Tuple7.class, Tuple8.class, Tuple9.class, Tuple10.class, Tuple11.class, Tuple12.class, Tuple13.class, Tuple14.class, Tuple15.class, Tuple16.class, Tuple17.class, Tuple18.class, Tuple19.class, Tuple20.class, Tuple21.class, Tuple22.class, Tuple23.class, Tuple24.class, Tuple25.class};

    public Tuple() {
    }

    public abstract <T> T getField(int var1);

    public <T> T getFieldNotNull(int pos) {
        T field = this.getField(pos);
        if (field != null) {
            return field;
        } else {
            throw new NullFieldException(pos);
        }
    }

    public abstract <T> void setField(T var1, int var2);

    public abstract int getArity();

    public abstract <T extends Tuple> T copy();

    public static Class<? extends Tuple> getTupleClass(int arity) {
        if (arity >= 0 && arity <= 25) {
            return CLASSES[arity];
        } else {
            throw new IllegalArgumentException("The tuple arity must be in [0, 25].");
        }
    }

    public static Tuple newInstance(int arity) {
        switch(arity) {
        case 0:
            return Tuple0.INSTANCE;
        case 1:
            return new Tuple1();
        case 2:
            return new Tuple2();
        case 3:
            return new Tuple3();
        case 4:
            return new Tuple4();
        case 5:
            return new Tuple5();
        case 6:
            return new Tuple6();
        case 7:
            return new Tuple7();
        case 8:
            return new Tuple8();
        case 9:
            return new Tuple9();
        case 10:
            return new Tuple10();
        case 11:
            return new Tuple11();
        case 12:
            return new Tuple12();
        case 13:
            return new Tuple13();
        case 14:
            return new Tuple14();
        case 15:
            return new Tuple15();
        case 16:
            return new Tuple16();
        case 17:
            return new Tuple17();
        case 18:
            return new Tuple18();
        case 19:
            return new Tuple19();
        case 20:
            return new Tuple20();
        case 21:
            return new Tuple21();
        case 22:
            return new Tuple22();
        case 23:
            return new Tuple23();
        case 24:
            return new Tuple24();
        case 25:
            return new Tuple25();
        default:
            throw new IllegalArgumentException("The tuple arity must be in [0, 25].");
        }
    }
}

查看源碼我們看到Tuple0一直到Tuple25

我們看flink為我們?yōu)槲覀儤?gòu)造好了0-25個(gè)字段的模板類(lèi)

ackage org.apache.flink.api.java.tuple;

import java.io.ObjectStreamException;
import org.apache.flink.annotation.Public;

@Public
public class Tuple0 extends Tuple {
    private static final long serialVersionUID = 1L;
    public static final Tuple0 INSTANCE = new Tuple0();

    public Tuple0() {
    }

    public int getArity() {
        return 0;
    }

    public <T> T getField(int pos) {
        throw new IndexOutOfBoundsException(String.valueOf(pos));
    }

    public <T> void setField(T value, int pos) {
        throw new IndexOutOfBoundsException(String.valueOf(pos));
    }

    public Tuple0 copy() {
        return new Tuple0();
    }

    public String toString() {
        return "()";
    }

    public boolean equals(Object o) {
        return this == o || o instanceof Tuple0;
    }

    public int hashCode() {
        return 0;
    }

    private Object readResolve() throws ObjectStreamException {
        return INSTANCE;
    }
}

三、Tuple的使用

方式一：初始化元組

可使用靜態(tài)方法 newInstance進(jìn)行元組構(gòu)造指定元組空間大小;

ex: 1 則元組只有一個(gè)空間，則實(shí)際使用的Tuple1 字段只有f0

ex: 12 則元組只有兩個(gè)空間，則實(shí)際使用的Tuple2 字段只有f0,f1

指定 Tuple元組空間大小（可理解為字段個(gè)數(shù)）

Tuple tuple = Tuple.newInstance(1);

方式一：構(gòu)造元組

使用Tuple.newInstance(xx),指定元組空間大小的話，這樣存取雖然能夠?qū)崿F(xiàn)，但會(huì)存在存儲(chǔ)索引位置使用不正確的情況，可能由于失誤操作編寫(xiě)出索引越界異常，而且使用不太方便，使用Tuplex.of(數(shù)據(jù))方法構(gòu)造Tuple元組

Tuple3<String, String, String> tuple3 = Tuple3.of("test0", "test1", "test2");
System.out.println(tuple3.f0); // test0
System.out.println(tuple3.f1); // test1
System.out.println(tuple3.f2); // test2

四、Flink之POJO類(lèi)型

Java和Scala的類(lèi)在滿足下列條件時(shí)，將會(huì)被Flink視作特殊的POJO數(shù)據(jù)類(lèi)型專(zhuān)門(mén)進(jìn)行處理：

1.是公共類(lèi)；

2.無(wú)參構(gòu)造是公共的；

3.所有的屬性都是可獲得的（聲明為公共的，或提供get,set方法）；

4.字段的類(lèi)型必須是Flink支持的。Flink會(huì)用Avro來(lái)序列化任意的對(duì)象。

Flink會(huì)分析POJO類(lèi)型的結(jié)構(gòu)獲知POJO的字段。POJO類(lèi)型要比一般類(lèi)型好用。此外，F(xiàn)link訪問(wèn)POJO要比一般類(lèi)型更高效。

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("word");

五、Flink之基本類(lèi)型

Flink支持Java和Scala所有的基本數(shù)據(jù)類(lèi)型，比如 Integer，String,和Double。

六、Flink之通用類(lèi)型

Flink支持大多數(shù)的Java,Scala類(lèi)（API和自定義）。包含不能序列化字段的類(lèi)在增加一些限制后也可支持。遵循Java Bean規(guī)范的類(lèi)一般都可以使用。

所有不能視為POJO的類(lèi)Flink都會(huì)當(dāng)做一般類(lèi)處理。這些數(shù)據(jù)類(lèi)型被視作黑箱，其內(nèi)容是不可見(jiàn)的。通用類(lèi)使用Kryo進(jìn)行序列/反序列化。

七、Flink之值類(lèi)型Values

通過(guò)實(shí)現(xiàn)org.apache.flinktypes.Value接口的read和write方法提供自定義代碼來(lái)進(jìn)行序列化/反序列化，而不是使用通用的序列化框架。

Flink預(yù)定義的值類(lèi)型與原生數(shù)據(jù)類(lèi)型是一一對(duì)應(yīng)的(例如:ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue)。這些值類(lèi)型作為原生數(shù)據(jù)類(lèi)型的可變變體，他們的值是可以改變的，允許程序重用對(duì)象從而緩解GC的壓力。

八、Flink之Hadoop的Writable類(lèi)

它實(shí)現(xiàn)org.apache.hadoop.Writable接口的類(lèi)型，該類(lèi)型的序列化邏輯在write()和readFields()方法中實(shí)現(xiàn)。

九、Flink之特殊類(lèi)型

Flink比較特殊的類(lèi)型有以下兩種：

1.Scala的 Either、Option和Try。

2.Java ApI有自己的Either實(shí)現(xiàn)。

Java Api 與 Scala 的類(lèi)似Either，它表示兩種可能類(lèi)型的值，Left或Right。Either對(duì)于錯(cuò)誤處理或需要輸出兩種不同類(lèi)型的記錄的運(yùn)算符很有用。

類(lèi)型擦除和類(lèi)型推理

Java編譯器在編譯之后會(huì)丟棄很多泛型類(lèi)型信息。這在Java中稱(chēng)為類(lèi)型擦除。這意味著在運(yùn)行時(shí)，對(duì)象的實(shí)例不再知道其泛型類(lèi)型。

例如，在JVM中，DataStream<String>和DataStream<Long>的實(shí)例看起來(lái)是相同的。

List<String> l1 = new ArrayList<String>();
List<Integer> l2 = new ArrayList<Integer>();
System.out.println(l1.getClass() == l2.getClass());

泛型：一種較為準(zhǔn)確的說(shuō)法就是為了參數(shù)化類(lèi)型，或者說(shuō)可以將類(lèi)型當(dāng)作參數(shù)傳遞給一個(gè)類(lèi)或者是方法。Flink 的Java API會(huì)試圖去重建（可以做類(lèi)型推理）這些被丟棄的類(lèi)型信息，并將它們明確地存儲(chǔ)在數(shù)據(jù)集以及操作中。你可以通過(guò)DataStream.getType()方法來(lái)獲取類(lèi)型，這個(gè)方法將返回一個(gè)TypeInformation的實(shí)例，這個(gè)實(shí)例是Flink內(nèi)部表示類(lèi)型的方式。

到此這篇關(guān)于Flink支持哪些數(shù)據(jù)類(lèi)型?的文章就介紹到這了,更多相關(guān)Flink的數(shù)據(jù)類(lèi)型內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: