Apache下Flink transformation的用法

这篇文章主要介绍"Apache下Flink transformation的用法"，在日常操作中，相信很多人在Apache下Flink transformation的用法问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"Apache下Flink transformation的用法"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

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))  }}

这里的数据源是一个1到10的list集合。Map的原理是：假设data数据集中有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()

输出结果：

234567891011

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的泛型是，第一个参数表示输入的类型，第二个参数表示输出是一个Integer类型。

Filter Function

将每个元素执行+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有什么区别？

需求：DataSource 中有100个元素,把结果存储在数据库中

如果使用map function ，那么实现方法如下：

  // DataSource 中有100个元素,把结果存储在数据库中  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=>{      // 每一个元素要存储到数据库中去，肯定需要先获取到connection      val connection = DBUtils.getConnection()      println(connection + " ... ")      // TODO .... 保存数据到DB      DBUtils.returnConnection(connection)    }).print()  }

打印结果，将会打印100个获取DBUtils.getConnection()的请求。如果数据量增多，显然不停的获取连接是不现实的。

因此MapPartition就应运而生了，转换一个分区里面的数据，也就是说一个分区中的数据调用一次。

因此要首先设置分区：

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

设置4个分区，也就是并行度，然后使用mapPartition来处理：

data.mapPartition(x => {      val connection = DBUtils.getConnection()      println(connection + " ... ")      // TODO .... 保存数据到DB      DBUtils.returnConnection(connection)      x    }).print()

那么就会的到4次连接请求，每一个分区获取一个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