概述

flink的sink是flink三大逻辑结构之一(source，transform，sink),功能就是负责把flink处理后的数据输出到外部系统中，flink 的sink和source的代码结构类似。

在编写代码的过程中，我们可以使用flink已经提供的sink，如kafka，jdbc,es等，当然我们也可以通过自定义的方式，来实现我们自己的sink。下面说明核心类

核心类

SinkFunction 是一个接口，类似于SourceFunction接口。SinkFunction中主要包含一个方法，那就是用于数据输出的invoke 方法,每条记录都会执行一次invoke方法，用于执行输出操作。

// Writes the given value to the sink. This function is called for every record.default void invoke(IN value) throws Exception

default void invoke(IN value, Context context) throws Exception

// Context接口中返回关于时间的信息interface Context {

/** Returns the current processing time. */

long currentProcessingTime();

/** Returns the current event-time watermark. */

long currentWatermark();

/*** Returns the timestamp of the current input record or {@code null} if the element does not* have an assigned timestamp.*/

Long timestamp();

}

我们一般自定义Sink的时候，都是继承AbstractRichFunction，他是一个抽象类，实现了RichFunction接口。

public abstract class AbstractRichFunction implements RichFunction, Serializable

并且提供了关于RuntimContext的操作和open，clone方法。

AbstractRichFunction 有很多实现类，比如关于msyql操作的JDBCSinkFunction，比如直接输出结果的 PrintSinkFunction，在我们开发的过程中，我们进程用print语句来打印结果，但是print函数中就是讲PrintSinkFunction类传递到addSink方法中。

public DataStreamSink print() {

PrintSinkFunction printFunction = new PrintSinkFunction<>();

return addSink(printFunction).name("Print to Std. Out");

}

PrintSinkFunction

我们这里分析一下PrintSinkFunction这个类，这个类就是将没个元素输出到标准输出或者是标准错误输出流中。

public class PrintSinkFunction extends RichSinkFunction {

private static final long serialVersionUID = 1L;

private final PrintSinkOutputWriter writer;

/*** Instantiates a print sink function that prints to standard out.*/

public PrintSinkFunction() {

writer = new PrintSinkOutputWriter<>(false);

}

/*** Instantiates a print sink function that prints to standard out.** @param stdErr True, if the format should print to standard error instead of standard out.*/

public PrintSinkFunction(final boolean stdErr) {

writer = new PrintSinkOutputWriter<>(stdErr);

}

/*** Instantiates a print sink function that prints to standard out and gives a sink identifier.** @param stdErr True, if the format should print to standard error instead of standard out.* @param sinkIdentifier Message that identify sink and is prefixed to the output of the value*/

public PrintSinkFunction(final String sinkIdentifier, final boolean stdErr) {

writer = new PrintSinkOutputWriter<>(sinkIdentifier, stdErr);

}

@Override

public void open(Configuration parameters) throws Exception {

super.open(parameters);

StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();

writer.open(context.getIndexOfThisSubtask(), context.getNumberOfParallelSubtasks());

}

@Override

public void invoke(IN record) {

writer.write(record);

}

@Override

public String toString() {

return writer.toString();

}

}

分析：

1、调用构造函数来创建一个PrintSinkOutputWriter

2、调用open方法中在调用PrintSinkOutputWriter 的open方法，进行初始化

3、调用invoke方法，通过PrintSinkOutputWriter 的writer方法吧record输出

自定义sink

我们这里自定义一个msyql的sink，也就是把flink中的数据，最后输出到mysql中。

public class MyMysqlSink extends RichSinkFunction {

private PreparedStatement ps = null;

private Connection connection = null;

String driver = "com.mysql.jdbc.Driver";

String url = "jdbc:mysql://127.0.0.1:3306/flinkdb";

String username = "root";

String password = "root";

// 初始化方法 @Override

public void open(Configuration parameters) throws Exception {

super.open(parameters);

// 获取连接 connection = getConn();

//执行查询 ps = connection.prepareStatement("select * from person;");

}

private Connection getConn() {

try {

Class.forName(driver);

connection = DriverManager.getConnection(url, username, password);

} catch (Exception e) {

e.printStackTrace();

}

return connection;

}

//Writes the given value to the sink. This function is called for every record. //每一个元素的插入，都会被调用一次invoke方法 @Override

public void invoke(Person p, Context context) throws Exception {

ps.setString(1,p.getName());

ps.setInt(2,p.getAge());

ps.executeUpdate();

}

@Override

public void close() throws Exception {

super.close();

if(connection != null){

connection.close();

}

if (ps != null){

ps.close();

}

}

}

程序调用入口

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 添加自定义数据源 DataStreamSource data = env.addSource(new MyMysqlSource());

data.print().setParallelism(2);

data.addSink(new MyMysqlSink());

// 提交执行任务env.execute("MySourceMysql");