flink sink

flink sink,第1张

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 中

 

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zaji/5136507.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-11-17
下一篇 2022-11-17

发表评论

登录后才能评论

评论列表(0条)

保存