Mongodb作为Flink 的 Sink

MongoSink.Java

继承RichSinkFunction，重写open 和 invoke方法，还有close。

packagecom.e.firsh.spb.sink;importcom.alibaba.fastjson.JSON;importcom.e.firs.spb.utils.Mongo.MongoService;importorg.apache.flink.configuration.Configuration;importorg.apache.flink.streaming.api.functions.sink.RichSinkFunction;importjava.net.UnknownHostException;/**
* Created by zhangjianxin on 2017/7/31.
*/publicclassMongoSinkextendsRichSinkFunction<String>{publicstaticStringCollectionName="collection-a";privateMongoServicemongoService;@Overridepublicvoidinvoke(Stringt){try{this.mongoService.saveJson(JSON.parseObject(t),CollectionName);}catch(UnknownHostExceptione){e.printStackTrace();}}@Overridepublicvoidopen(Configurationconfig){mongoService=newMongoService();try{super.open(config);}catch(Exceptione){e.printStackTrace();}}};