這篇文章主要介紹“Apache下Flink transformation的用法”，在日常操作中，相信很多人在Apache下Flink transformation的用法問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Apache下Flink transformation的用法”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

10年積累的成都網(wǎng)站設(shè)計、成都做網(wǎng)站、外貿(mào)網(wǎng)站建設(shè)經(jīng)驗，可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你，你也不認(rèn)識我。但先網(wǎng)站設(shè)計制作后付款的網(wǎng)站建設(shè)流程，更有崖州免費網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。

Map Function

Scala

新建一個Object

object DataSetTransformationApp {

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

  }

  def mapFunction(env: ExecutionEnvironment): Unit = {
    val data = env.fromCollection(List(1,2,3,4,5,6,7,8,9,10))
  }

}

這里的數(shù)據(jù)源是一個1到10的list集合。Map的原理是：假設(shè)data數(shù)據(jù)集中有N個元素，將每一個元素進行轉(zhuǎn)化：

data.map { x => x.toInt }

好比：y=f(x)

    // 對data中的每一個元素都去做一個+1操作
    data.map((x:Int) => x + 1 ).print()

然后對每一個元素都做一個+1操作。

簡單寫法：

如果這個里面只有一個元素，就可以直接寫成下面形式：

data.map((x) => x + 1).print()

更簡潔的寫法：

data.map(x => x + 1).print()

更簡潔的方法：

data.map(_ + 1).print()

輸出結(jié)果：

2
3
4
5
6
7
8
9
10
11

Java

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        mapFunction(executionEnvironment);
    }

    public static void mapFunction(ExecutionEnvironment executionEnvironment) throws Exception {
        List list = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            list.add(i + "");
        }
        DataSource data = executionEnvironment.fromCollection(list);
        data.map(new MapFunction() {
            public Integer map(String input) {
                return Integer.parseInt(input) + 1;
            }
        }).print();
    }

因為我們定義的List是一個String的泛型，因此MapFunction的泛型是，第一個參數(shù)表示輸入的類型，第二個參數(shù)表示輸出是一個Integer類型。

Filter Function

將每個元素執(zhí)行+1操作，并取出大于5的元素。

Scala

  def filterFunction(env: ExecutionEnvironment): Unit = {
    val data = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    data.map(_ + 1).filter(_ > 5).print()
  }

filter只會返回滿足條件的記錄。

Java

    public static void filterFunction(ExecutionEnvironment env) throws Exception {
        List list = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            list.add(i);
        }
        DataSource data = env.fromCollection(list);
        data.map(new MapFunction() {
            public Integer map(Integer input) {
                return input + 1;
            }
        }).filter(new FilterFunction() {
            @Override
            public boolean filter(Integer input) throws Exception {
                return input > 5;
            }
        }).print();
    }

MapPartition Function

map function 與 MapPartition function有什么區(qū)別？

需求：DataSource 中有100個元素,把結(jié)果存儲在數(shù)據(jù)庫中

如果使用map function ，那么實現(xiàn)方法如下：

  // DataSource 中有100個元素,把結(jié)果存儲在數(shù)據(jù)庫中
  def mapPartitionFunction(env: ExecutionEnvironment): Unit = {
    val students = new ListBuffer[String]
    for (i <- 1 to 100) {
      students.append("Student" + i)
    }
    val data = env.fromCollection(students)
    data.map(x=>{
      // 每一個元素要存儲到數(shù)據(jù)庫中去，肯定需要先獲取到connection
      val connection = DBUtils.getConnection()
      println(connection + " ... ")
      // TODO .... 保存數(shù)據(jù)到DB
      DBUtils.returnConnection(connection)
    }).print()
  }

打印結(jié)果，將會打印100個獲取DBUtils.getConnection()的請求。如果數(shù)據(jù)量增多，顯然不停的獲取連接是不現(xiàn)實的。

因此MapPartition就應(yīng)運而生了，轉(zhuǎn)換一個分區(qū)里面的數(shù)據(jù)，也就是說一個分區(qū)中的數(shù)據(jù)調(diào)用一次。

因此要首先設(shè)置分區(qū)：

val data = env.fromCollection(students).setParallelism(4)

設(shè)置4個分區(qū)，也就是并行度，然后使用mapPartition來處理：

data.mapPartition(x => {
      val connection = DBUtils.getConnection()
      println(connection + " ... ")
      // TODO .... 保存數(shù)據(jù)到DB
      DBUtils.returnConnection(connection)
      x
    }).print()

那么就會的到4次連接請求，每一個分區(qū)獲取一個connection。

Java

public static void mapPartitionFunction(ExecutionEnvironment env) throws Exception {
        List list = new ArrayList<>();
        for (int i = 1; i <= 100; i++) {
            list.add("student:" + i);
        }
        DataSource data = env.fromCollection(list);
        /*data.map(new MapFunction() {
            @Override
            public String map(String input) throws Exception {
                String connection = DBUtils.getConnection();
                System.out.println("connection = [" + connection + "]");
                DBUtils.returnConnection(connection);
                return input;
            }
        }).print();*/
        data.mapPartition(new MapPartitionFunction() {
            @Override
            public void mapPartition(Iterable values, Collector