4.1.13 Flink-流处理框架-Flink流处理API之实现 UDF 函数（更细粒度的控制流）

超、凢脫俗 2022-11-04 11:15 125阅读 0赞

**目录**

1.函数类（Function Classes）

补充小知识：静态内部类

2.匿名函数（Lambda Functions）

3.富函数（Rich Functions）

--------------------

# 1.函数类（Function Classes） #

Flink 暴露了所有 udf 函数的接口(实现方式为接口或者抽象类)。例如 MapFunction, FilterFunction, ProcessFunction 等等。

下面例子实现了 FilterFunction 接口：

DataStream<String> flinkTweets = tweets.filter(new FlinkFilter());
    
    public static class FlinkFilter implements FilterFunction<String> {
        @Override    
        public boolean filter(String value) throws Exception {
            return value.contains("flink");
        }
    }

还可以将函数实现成匿名类

DataStream<String> flinkTweets = tweets.filter(new FilterFunction<String>() {    
        @Override
        public boolean filter(String value) throws Exception {
            return value.contains("flink");
        }
    });

我们 filter 的字符串"flink"还可以当作参数传进去。

DataStream<String> tweets = env.readTextFile("INPUT_FILE ");
    
    DataStream<String> flinkTweets = tweets.filter(new KeyWordFilter("flink"));
    
    public static class KeyWordFilter implements FilterFunction<String> {
            private String keyWord;
            KeyWordFilter(String keyWord) { this.keyWord = keyWord; }
            
            @Override
            public boolean filter(String value) throws Exception {
                return value.contains(this.keyWord);
            }
    }

##         补充小知识：静态内部类 ##

java允许我们在一个类里面定义静态类。比如内部类（nested class）。把nested class封闭起来的类叫外部类。在java中，我们不能用static修饰顶级类（top level class）。只有内部类可以为static。静态内部类和非静态内部类之间到底有什么不同呢？下面是**两者间主要的不同**。

*      （1）静态内部类跟静态方法一样，只能访问静态的成员变量和方法，不能访问非静态的方法和属性，但是普通内部类可以访问任意外部类的成员变量和方法；
 *      （2）静态内部类可以声明普通成员变量和方法，而普通内部类不能声明static成员变量和方法；
 *      （3）一个非静态内部类不能脱离外部类实体被创建，一个非静态内部类可以访问外部类的数据和方法，因为他就在外部类里面。

静态内部类使用场景一般是**当外部类需要使用内部类，而内部类无需外部类资源，并且内部类可以单独创建的时候**会考虑采用静态内部类的设计

# 2.匿名函数（Lambda Functions） #

使用匿名函数方式，就不能将flink当成一个keyword传参进来了，这个时候只能固定写法。

DataStream<String> tweets = env.readTextFile("INPUT_FILE");
    DataStream<String> flinkTweets = tweets.filter( tweet -> tweet.contains("flink") );

# 3.富函数（Rich Functions） #

“富函数”是 DataStream API 提供的一个函数类的接口，**所有 Flink 函数类都 有其 Rich 版本**。它与常规函数的不同在于，**可以获取运行环境的上下文，并拥有一些生命周期方法**，所以可以实现更复杂的功能。⚫ RichMapFunction ⚫ RichFlatMapFunction ⚫ RichFilterFunction ⚫ …

Rich Function 有一个生命周期的概念。典型的生命周期方法有：

*  ⚫ open()方法是 rich function 的初始化方法，当一个算子例如 map 或者 filter 被调用之前 open()会被调用。
 *  ⚫ close()方法是生命周期中的最后一个调用的方法，做一些清理工作。
 *  ⚫ getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息，例如函 数执行的并行度，任务的名字，以及 state 状态

public static class MyMapFunction extends RichMapFunction<SensorReading,Tuple2<Integer, String>> {
        @Override
        public Tuple2<Integer, String> map(SensorReading value) throws Exception {
            return new Tuple2<>(getRuntimeContext().getIndexOfThisSubtask(),value.getId());
        }
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("my map open");
            // 以下可以做一些初始化工作，例如建立一个和 HDFS 的连接
        }
        @Override
        public void close() throws Exception {
             System.out.println("my map close");
            // 以下做一些清理工作，例如断开和 HDFS 的连接
        }
    }

package com.atguigu.apitest.transform;
    
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.functions.RichMapFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    
    public class TransformTest5_RichFunction {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(4);
    
            // 从文件读取数据
            DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            DataStream<Tuple2<String, Integer>> resultStream = dataStream.map( new MyMapper() );
    
            resultStream.print();
    
            env.execute();
        }
    
        public static class MyMapper0 implements MapFunction<SensorReading, Tuple2<String, Integer>>{
            @Override
            public Tuple2<String, Integer> map(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), value.getId().length());
            }
        }
    
        // 实现自定义富函数类
        public static class MyMapper extends RichMapFunction<SensorReading, Tuple2<String, Integer>>{
            @Override
            public Tuple2<String, Integer> map(SensorReading value) throws Exception {
    //            getRuntimeContext().getState();
                return new Tuple2<>(value.getId(), getRuntimeContext().getIndexOfThisSubtask());
            }
    
            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化工作，一般是定义状态，或者建立数据库连接
                System.out.println("open");
            }
    
            @Override
            public void close() throws Exception {
                // 一般是关闭连接和清空状态的收尾操作
                System.out.println("close");
            }
        }
    }