本篇文章給大家分享的是有關(guān)如何實(shí)現(xiàn)Apache Flink中Flink數(shù)據(jù)流轉(zhuǎn)換，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說(shuō)，跟著小編一起來(lái)看看吧。

濱湖網(wǎng)站制作公司哪家好，找成都創(chuàng)新互聯(lián)公司！從網(wǎng)頁(yè)設(shè)計(jì)、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、成都響應(yīng)式網(wǎng)站建設(shè)等網(wǎng)站項(xiàng)目制作，到程序開發(fā)，運(yùn)營(yíng)維護(hù)。成都創(chuàng)新互聯(lián)公司從2013年成立到現(xiàn)在10年的時(shí)間，我們擁有了豐富的建站經(jīng)驗(yàn)和運(yùn)維經(jīng)驗(yàn)，來(lái)保證我們的工作的順利進(jìn)行。專注于網(wǎng)站建設(shè)就選成都創(chuàng)新互聯(lián)公司。

Operators操作轉(zhuǎn)換一個(gè)或多個(gè)DataStream到一個(gè)新的DataStream 。

filter function

Scala

object DataStreamTransformationApp {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    filterFunction(env)
    env.execute("DataStreamTransformationApp")
  }

  def filterFunction(env: StreamExecutionEnvironment): Unit = {
    val data=env.addSource(new CustomNonParallelSourceFunction)
    data.map(x=>{
      println("received:" + x)
      x
    }).filter(_%2 == 0).print().setParallelism(1)
  }

}

數(shù)據(jù)源選擇之前的任意一個(gè)數(shù)據(jù)源即可。

這里的map中沒(méi)有做任何實(shí)質(zhì)性的操作，filter中將所有的數(shù)都對(duì)2取模操作，打印結(jié)果如下:

received:1
received:2
2
received:3
received:4
4
received:5
received:6
6
received:7
received:8
8

說(shuō)明map中得到的所有的數(shù)據(jù)，而在filter中進(jìn)行了過(guò)濾操作。

Java

    public static void filterFunction(StreamExecutionEnvironment env) {
        DataStreamSource<Long> data = env.addSource(new JavaCustomParallelSourceFunction());
        data.setParallelism(1).map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                System.out.println("received:"+value);
                return value;
            }
        }).filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return value % 2==0;
            }
        }).print().setParallelism(1);
    }

需要先使用data.setParallelism(1)然后再進(jìn)行map操作，否則會(huì)輸出多次。因?yàn)槲覀冇玫氖荍avaCustomParallelSourceFunction()，而當(dāng)我們使用JavaCustomNonParallelSourceFunction時(shí)，默認(rèn)就是并行度1，可以不用設(shè)置。

Union Function

Scala

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

//    filterFunction(env)
    unionFunction(env)
    env.execute("DataStreamTransformationApp")
  }

  def unionFunction(env: StreamExecutionEnvironment): Unit = {
    val data01 = env.addSource(new CustomNonParallelSourceFunction)
    val data02 = env.addSource(new CustomNonParallelSourceFunction)
    data01.union(data02).print().setParallelism(1)

  }

Union操作將兩個(gè)數(shù)據(jù)集綜合起來(lái)，可以一同處理，上面打印輸出如下：

1
1
2
2
3
3
4
4

Java

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
//        filterFunction(environment);
        unionFunction(environment);
        environment.execute("JavaDataStreamTransformationApp");
    }

    public static void unionFunction(StreamExecutionEnvironment env) {
        DataStreamSource<Long> data1 = env.addSource(new JavaCustomNonParallelSourceFunction());
        DataStreamSource<Long> data2 = env.addSource(new JavaCustomNonParallelSourceFunction());
        data1.union(data2).print().setParallelism(1);
    }

Split  Select  Function

Scala

split可以將一個(gè)流拆成多個(gè)流，select可以從多個(gè)流中進(jìn)行選擇處理的流。

def splitSelectFunction(env: StreamExecutionEnvironment): Unit = {
    val data = env.addSource(new CustomNonParallelSourceFunction)
    val split = data.split(new OutputSelector[Long] {
      override def select(value: Long): lang.Iterable[String] = {
        val list = new util.ArrayList[String]()
        if (value % 2 == 0) {
          list.add("even")
        } else {
          list.add("odd")
        }
        list
      }
    })
    split.select("odd","even").print().setParallelism(1)
  }

可以根據(jù)選擇的名稱來(lái)處理數(shù)據(jù)。

Java

public static void splitSelectFunction(StreamExecutionEnvironment env) {
        DataStreamSource<Long> data = env.addSource(new JavaCustomNonParallelSourceFunction());
        SplitStream<Long> split = data.split(new OutputSelector<Long>() {
            @Override
            public Iterable<String> select(Long value) {
                List<String> output = new ArrayList<>();
                if (value % 2 == 0) {
                    output.add("odd");
                } else {
                    output.add("even");
                }
                return output;
            }
        });
        split.select("odd").print().setParallelism(1);
    }

以上就是如何實(shí)現(xiàn)Apache Flink中Flink數(shù)據(jù)流轉(zhuǎn)換，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見到或用到的。希望你能通過(guò)這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

新聞標(biāo)題：如何實(shí)現(xiàn)ApacheFlink中Flink數(shù)據(jù)流轉(zhuǎn)換
本文地址：http://jinyejixie.com/article6/ghhhog.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供自適應(yīng)網(wǎng)站、搜索引擎優(yōu)化、網(wǎng)站制作、品牌網(wǎng)站設(shè)計(jì)、云服務(wù)器、網(wǎng)站維護(hù)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)