Flink学习笔记(5)——DataSteam API

一、 DataStream API（基础篇）

1.1 执行环境（Execution Environment）

1.1.1 创建执行环境

1.1.2 执行模式(Execution Mode)

1.1.3 触发程序执行

1.2 源算子（Source）

1.2.1 准备工作

1.2.2 三种读取数据的方法(从集合中读取数据、从文件读取数据、从元素读取数据)

1.3 转换算子（Transformation）

1.3.1 基本转换算子

1.3.2 聚合算子（Aggregation）

1.3.3 用户自定义函数（UDF）

1.3.4 物理分区（Physical Partitioning）

1.3.4 输出到 Elasticsearch

1.4.6 输出到 MySQL（JDBC）

1.4.7 自定义 Sink 输出

1.5 本章总结

一、 DataStream API（基础篇）

1.1 执行环境（Execution Environment）

Flink 程序可以在各种上下文环境中运行：我们可以在本地 JVM 中执行程序，也可以提交到远程集群上运行。

不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前 Flink 的运行环境，从而建立起与 Flink 框架之间的联系。只有获取了环境上下文信息，才能将具体的任务调度到不同的 TaskManager 执行。

1.1.1 创建执行环境

编写Flink程序的第一步，就是创建执行环境。我们要获取的执行环境，是StreamExecutionEnvironment 类的对象，这是所有 Flink 程序的基础。在代码中创建执行环境的方式，就是调用这个类的静态方法，具体有以下三种。

1.1.2 执行模式(Execution Mode)

当然，在后面的示例代码中，即使是有界的数据源，我们也会统一用 STREAMING 模式处理。这是因为我们的主要目标还是构建实时处理流数据的程序，有界数据源也只是我们用来测试的手段。

1.1.3 触发程序执行

代码写完以后一定要加上execute方法才会触发真正的计算

1.2 源算子（Source）

1.2.1 准备工作


package com.me.chapter05;
 
import java.sql.Timestamp;
 
public class Event {
 
    public String user;
    public String url;
    public Long timestamp;
 
    public Event(){
 
 
    }
 
    public Event(String user, String url, Long timestamp) {
        this.user = user;
        this.url = url;
        this.timestamp = timestamp;
    }
 
    @Override
    public String toString() {
        return "Event{" +
                "user='" + user + '\'' +
                ", url='" + url + '\'' +
                ", timestamp=" + new Timestamp(timestamp) +
                '}';
    }
}

这里需要注意，我们定义的 Event，有这样几个特点：

⚫ 类是公有（public）的

⚫ 有一个无参的构造方法

⚫ 所有属性都是公有（public）的

⚫ 所有属性的类型都是可以序列化的 Flink 会把这样的类作为一种特殊的 POJO 数据类型来对待，方便数据的解析和序列化。另外我们在类中还重写了 toString 方法，主要是为了测试输出显示更清晰。关于 Flink 支持的数据类型，我们会在后面章节做详细说明。

我们这里自定义的 Event POJO 类会在后面的代码中频繁使用，所以在后面的代码中碰到

Event，把这里的 POJO 类导入就好了。

注：Java 编程比较好的实践是重写每一个类的 toString 方法

1.2.2 三种读取数据的方法(从集合中读取数据、从文件读取数据、从元素读取数据)


package com.me.chapter05;
 
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
import java.util.ArrayList;
 
public class SourceTest {
    public static void main(String[] args) throws Exception{
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        //1、从文件中读取数据(最常见)
        DataStreamSource stream1 = env.readTextFile("input/clicks.txt");
 
        //2、从集合中读取数据
        ArrayList nums = new ArrayList<>();
        nums.add(2);
        nums.add(5);
        DataStream numStream = env.fromCollection(nums);
 
        ArrayList events = new ArrayList<>();
        events.add(new Event("Mary","./home",1000L));
        events.add(new Event("Bob","./cart",2000L));
        DataStream stream2= env.fromCollection(events);
 
 
        //3、从元素读取数据
        DataStream stream3= env.fromElements(
                new Event("Mary","./home",1000L),
                new Event("Bob","./cart",2000L)
        );
 
        stream1.print("1");
        numStream.print("nums");
        stream2.print("2");
        stream3.print("3");
 
        env.execute();
        
    }
}

这三种方法都是对有界流的处理

1.2.3 从 Socket 读取数据

1.2.4 从 Kafka 读取数据


 
 org.apache.flink 
 flink-connector-kafka_${scala.binary.version} 
 ${flink.version}

然后调用 env.addSource()，传入 FlinkKafkaConsumer 的对象实例就可以了。


import org.apache.flink.api.common.serialization.SimpleStringSchema; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; 
 
import java.util.Properties; 
 
public class SourceKafkaTest { 
 public static void main(String[] args) throws Exception { 
 StreamExecutionEnvironment env = 
 StreamExecutionEnvironment.getExecutionEnvironment(); 
 env.setParallelism(1); 
 
 Properties properties = new Properties(); 
 properties.setProperty("bootstrap.servers", "hadoop102:9092"); 
 properties.setProperty("group.id", "consumer-group"); 
 properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); 
 properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); 
 properties.setProperty("auto.offset.reset", "latest"); 
 
 DataStreamSource stream = env.addSource(new FlinkKafkaConsumer( 
 "clicks", 
 new SimpleStringSchema(), 
 properties 
 )); 
 
 stream.print("Kafka"); 
 
 env.execute(); 
 } 
}

1.2.6 自定义Source


package com.me.chapter05;
 
import org.apache.flink.streaming.api.functions.source.SourceFunction;
 
import java.util.Calendar;
import java.util.Random;
 
 
public class ClickSource implements SourceFunction {
    //声明一个标志位
    private Boolean running=true;
 
    @Override
    public void run(SourceContext ctx) throws Exception {
        Random random = new Random(); // 在指定的数据集中随机选取数据
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};
 
        while (running) {
            ctx.collect(new Event(
                    users[random.nextInt(users.length)],
                    urls[random.nextInt(urls.length)],
                    Calendar.getInstance().getTimeInMillis()
            ));
            // 隔 1 秒生成一个点击事件，方便观测
            Thread.sleep(1000);
        }
    }
    @Override
    public void cancel() {
 
    }
}


package com.me.chapter05;
 
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
 
import java.util.Random;
 
public class SourceCustomTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        //这种简单的自定义Source方法只可以指定并行度为1
        //DataStreamSource customStream=env.addSource(new ClickSource());
 
        DataStreamSource customStream = env.addSource(new ParallelCustomSource()).setParallelism(2);
 
        customStream.print();
 
        env.execute();
    }
 
    //实现自定义的并行SourceFunction，这样可以指定更高的并行度(这里直接用静态类完成，不去重新新建一个类了)
    public static class ParallelCustomSource implements ParallelSourceFunction{
        private Boolean running=true;
        private Random random=new Random();
 
        @Override
        public void run(SourceContext ctx) throws Exception {
            while(running){
                ctx.collect(random.nextInt());
            }
        }
 
        @Override
        public void cancel() {
        }
    }
}

1.2.7 Flink 支持的数据类型

我们已经了解了 Flink 怎样从不同的来源读取数据。在之前的代码中，我们的数据都是定义好的 UserBehavior 类型，而且在之前小节中特意说明了对这个类的要求。那还有没有其他更灵活的类型可以用呢？Flink 支持的数据类型到底有哪些？

1.3 转换算子（Transformation）

1.3.1 基本转换算子


package com.me.chapter05;
 
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransformMapTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
 
        // 传入匿名类，实现 MapFunction
        stream.map(new MapFunction() {
            @Override
            public String map(Event e) throws Exception {
                return e.user;
            }
        });
 
        // 传入 MapFunction 的实现类
        stream.map(new UserExtractor()).print();
 
        env.execute();
    }
 
    //单独定义的一个静态类
    public static class UserExtractor implements MapFunction {
        @Override
        public String map(Event e) throws Exception {
            return e.user;
        }
    }
}

上面代码中，MapFunction 实现类的泛型类型，与输入数据类型和输出数据的类型有关。在实现 MapFunction 接口的时候，需要指定两个泛型，分别是输入事件和输出事件的类型，还需要重写一个 map()方法，定义从一个输入事件转换为另一个输出事件的具体逻辑。

另外，细心的读者通过查看 Flink 源码可以发现，基于 DataStream 调用 map 方法，返回的其实是一个 SingleOutputStreamOperator。

public  SingleOutputStreamOperator map(MapFunction mapper){}

这表示 map 是一个用户可以自定义的转换（transformation）算子，它作用于一条数据流上，转换处理的结果是一个确定的输出类型。当然，SingleOutputStreamOperator 类本身也继承自DataStream 类，所以说 map 是将一个 DataStream 转换成另一个 DataStream 是完全正确的。


package com.me.chapter05;
 
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransformFilterTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
 
        // 传入匿名类实现 FilterFunction
        stream.filter(new FilterFunction() {
            @Override
            public boolean filter(Event e) throws Exception {
                return e.user.equals("Mary");
            }
        });
 
        // 传入 FilterFunction 实现类
        stream.filter(new UserFilter()).print();
 
        env.execute();
    }
    public static class UserFilter implements FilterFunction {
        @Override
        public boolean filter(Event e) throws Exception {
            return e.user.equals("Mary");
        }
    }
}


package com.me.chapter05;
 
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
import org.apache.flink.util.Collector;
 
public class TransformFlatMapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
 
        stream.flatMap(new MyFlatMap()).print();
 
        env.execute();
    }
    public static class MyFlatMap implements FlatMapFunction {
        @Override
        public void flatMap(Event value, Collector out) throws Exception
        {
            if (value.user.equals("Mary")) {
                out.collect(value.user);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.url);
            }
        }
    }
}

1.3.2 聚合算子（Aggregation）


package com.me.chapter05;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransformSimpleTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
 
        // 使用 Lambda 表达式
        KeyedStream keyedStream = stream.keyBy(e -> e.user);
 
        // 使用匿名类实现 KeySelector
        KeyedStream keyedStream1 = stream.keyBy(new KeySelector() {
            @Override
            public String getKey(Event e) throws Exception {
                return e.user;
            }
        });
 
        env.execute();
    }
}

例如，下面就是对元组数据流进行聚合的测试：


package com.me.chapter05;
 
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransTupleAggreationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource> stream = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("a", 3),
                Tuple2.of("b", 3),
                Tuple2.of("b", 4)
        );
 
        stream.keyBy(r -> r.f0).sum(1).print();
        stream.keyBy(r -> r.f0).sum("f1").print();
        stream.keyBy(r -> r.f0).max(1).print();
        stream.keyBy(r -> r.f0).max("f1").print();
        stream.keyBy(r -> r.f0).min(1).print();
        stream.keyBy(r -> r.f0).min("f1").print();
        stream.keyBy(r -> r.f0).maxBy(1).print();
        stream.keyBy(r -> r.f0).maxBy("f1").print();
        stream.keyBy(r -> r.f0).minBy(1).print();
        stream.keyBy(r -> r.f0).minBy("f1").print();
 
        env.execute();
    }
}

而如果数据流的类型是 POJO 类，那么就只能通过字段名称来指定，不能通过位置来指定了。


package com.me.chapter05;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransPojoAggregationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        DataStreamSource stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
 
        stream.keyBy(e -> e.user).max("timestamp").print(); // 指定字段名称
 
        env.execute();
    }
}


package com.me.chapter05;
 
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class TransReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 这里的 ClickSource()使用了之前自定义数据源小节中的 ClickSource()
        env.addSource(new ClickSource())
                // 将 Event 数据类型转换成元组类型
                .map(new MapFunction>() {
                    @Override
                    public Tuple2 map(Event e) throws Exception {
                        return Tuple2.of(e.user, 1L);
                    }
                })
                .keyBy(r -> r.f0) // 使用用户名来进行分流
                .reduce(new ReduceFunction>() {
                    @Override
                    public Tuple2 reduce(Tuple2 value1,
                                                       Tuple2 value2) throws Exception {
                        // 每到一条数据，用户 pv 的统计值加 1
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .keyBy(r -> true) // 为每一条数据分配同一个 key，将聚合结果发送到一条流中 去
                .reduce(new ReduceFunction>() {
                    @Override
                    public Tuple2 reduce(Tuple2 value1,
                                                       Tuple2 value2) throws Exception {
                        // 将累加器更新为当前最大的 pv 统计值，然后向下游发送累加器的值
                        return value1.f1 > value2.f1 ? value1 : value2;
                    }
                }).print();
        env.execute();
    }
}

reduce 同简单聚合算子一样，也要针对每一个 key 保存状态。因为状态不会清空，所以我们需要将 reduce 算子作用在一个有限 key 的流上。