flink sink

Sink（对外输出） Flink 没有类似于 spark 中 foreach 方法，让用户进行迭代的操作。虽有对外的 输出操作都要利用 Sink 完成。最后通过类似如下方式完成整个任务最终输出操作。 stream.addSink(new MySink(xxxx)) 官方提供了一部分的框架的 sink 。除此以外，需要用户自定义实现 sink 。

 

1 Kafka pom.xml

 主函数中添加 sink：

 2 Redis

 定义一个 redis 的 mapper 类，用于定义保存到 redis 时调用的命令：

 在主函数中调用：

 

 3 Elasticsearch

 在主函数中调用：

 ElasitcsearchSinkFunction 的实现：

 4 JDBC 自定义 sink

 添加 MyJdbcSink

 

 在 main 方法中增加，把明细保存到 mysql 中