Flink實(shí)時(shí)API-自定義DataSinkDataStream——自定義SinkDataStreamAPISourceTransformationsSinkDataStream——自定義SinkDataStreamAPISourceTransformationsSink數(shù)據(jù)接收器Sink將DataStream中的數(shù)轉(zhuǎn)發(fā)到文件，套接字，外部系統(tǒng)或打印它們。Connectors3）addSink：調(diào)用自定義接收器

Connectors即連接器，實(shí)現(xiàn)了某個(gè)組件的Source或Sink，我們可以使用官方內(nèi)置Connectors實(shí)現(xiàn)對(duì)數(shù)據(jù)流的實(shí)時(shí)讀取或?qū)崟r(shí)寫(xiě)出。官方內(nèi)置Connectors有如下：ApacheKafka(source/sink)Redis(sink)redis安裝與實(shí)踐步驟（1）下載并解壓文件redis-3.2.12.tar.gz到usr/local目錄下,重命名為redis$sudotar-zxf~/Downloads/redis-3.2.12.tar.gz-C/usr/local$sudomv./redis-3.2.12./redis#將文件夾名改為redisredis安裝與實(shí)踐（2）redis設(shè)置相應(yīng)權(quán)限$cd/usr/local$sudochown-Rhadoop./redis（3）對(duì)redis源碼進(jìn)行編譯$

cd/usr/local/redis/$make（4）修改redis配置文件redis.conf$

cd/usr/local/redis$geditredis.confredis安裝與實(shí)踐redis安裝與實(shí)踐（5）啟動(dòng)redis服務(wù)器端$cd/usr/local/redis/$src/redis-serverredis.confredis安裝與實(shí)踐（6）啟動(dòng)redis客戶端$cd/usr/local/redis/src/$./redis-cli（7）測(cè)試redis（8）關(guān)閉redis服務(wù)器端$cd/usr/local/redis/src/$./redis-clishutdownredisSink

FlinkKafka算子轉(zhuǎn)換操作數(shù)據(jù)源KafkaSourceRedisSinkredis寫(xiě)入redis

Sinkflink寫(xiě)入數(shù)據(jù)到Redis需要導(dǎo)入包寫(xiě)入步驟：①獲取redisSink②調(diào)用env.addSink(redisSink)redis

SinkobjectRedisUtil{valconf=newFlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build()defgetRedisSink()={//flinkJedisConfigBase規(guī)定數(shù)據(jù)寫(xiě)到哪兒,redisSinkMapper如何寫(xiě)，寫(xiě)何種數(shù)據(jù)類(lèi)型（string等）

newRedisSink[(String,Int)](conf,newRedisMapper[(String,Int)](){

overridedefgetCommandDescription:RedisCommandDescription=newRedisCommandDescription(RedisCommand.HSET,"wordcount")

overridedefgetKeyFromData(data:(String,Int)):String=data._1

overridedefgetValueFromData(data:(String,Int)):String=data._2.toString})}redis

SinkobjectRedisSinkDemo{defmain(args:Array[String]):Unit={valenv=StreamExecutionEnvironment.getExecutionEnvironment

//獲取DataSourcevallineStrDS=env.socketTextStream(hostname="localhost",port=8989)valds1=lineStrDS.flatMap(line=>line.split("")).map(word=>(word,1)).keyBy(_._1).s