Flink自定义source和sink，获取kafka的key，输出指定key

val kafkaSource = new FlinkKafkaConsumer[ObjectNode]("kafka_demo", new JsonNodeDeserializationSchema(), Common.getProp)
    val sink = new FlinkKafkaProducer[String]("kafka_demo_out", new SimpleStringSchema(), Common.getProp)
    sink.setWriteTimestampToKafka(true)

    env.addSource(kafkaSource)
      .map(node => {
        node.put("token", System.currentTimeMillis())
        node.toString
      })
      .addSink(sink)

object KafkaKeyMaker {
  val topic = "kafka_key"
  def main(args: Array[String]): Unit = {

    val producer = new KafkaProducer[String, String](Common.getProp)
    while (true) {
      val map = Map("user"->"venn", "name"->"venn","pass"->System.currentTimeMillis())
      val jsonObject: JSONObject = new JSONObject(map)
      println(jsonObject.toString())
      // key : msgKey + long
      val msg = new ProducerRecord[String, String](topic, "msgKey" + System.currentTimeMillis(), jsonObject.toString())
      producer.send(msg)
      producer.flush()
      Thread.sleep(3000)
    }

  }
}

object KafkaKeyReceive{
  val topic = "kafka_key"
  def main(args: Array[String]): Unit = {
    val consumer = new KafkaConsumer[String, String](Common.getProp)
    consumer.subscribe(util.Arrays.asList(topic + "_out"))
    while (true) {
      val records = consumer.poll(500)
      val tmp = records.iterator()
      while (tmp.hasNext){
        val record = tmp.next()
        val key = record.key()
        val value = record.value()
        println("receive -> key : " + key + ", value : " + value)
      }
      Thread.sleep(3000)
    }
  }
}

import com.venn.common.Common
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import scala.collection.JavaConversions._
/**
  * Created by venn on 19-4-26.
  */
object KafkaSourceKey {

  def main(args: Array[String]): Unit = {
    // environment
    val env: StreamExecutiOnEnvironment= StreamExecutionEnvironment.getExecutionEnvironment

    
    env.addSource(new RichSourceFunction[String] { // kafka consumer 对象 var consumer: KafkaConsumer[String, String] = null // 初始化方法 override def open(parameters: Configuration): Unit = { cOnsumer= new KafkaConsumer[String, String](Common.getProp) // 订阅topic val list = List("kafka_key") consumer.subscribe(list) } // 执行方法，拉取数据，获取到的数据，会放到source 的缓冲区 override def run(ctx: SourceFunction.SourceContext[String]): Unit = { println("run") while (true) { val records = consumer.poll(500) val tmp = records.iterator() while (tmp.hasNext) { val record = tmp.next() val key = record.key() val value = record.value() ctx.collect("key : " + key + ", value " + value) } } } override def cancel(): Unit = { println("cancel") } }).map(s => s + "map")
      .addSink(new RichSinkFunction[String] { // kafka producer 对象 var producer: KafkaProducer[String, String] = null // 初始化 override def open(parameters: Configuration): Unit = { producer = new KafkaProducer[String, String](Common.getProp) } override def close(): Unit = { if (producer == null) { producer.flush() producer.close() } } // 输出数据，每条结果都会执行一次，并发高的时候，可以按需做flush override def invoke(value: String, context: SinkFunction.Context[_]): Unit = { println("flink : " + value) val msg = new ProducerRecord[String, String]( "kafka_key_out", "key" + System.currentTimeMillis(), value) producer.send(msg) producer.flush() } }) // execute job
    env.execute("KafkaToKafka")
  }

}

{"user" : "venn", "name" : "venn", "pass" : 1561355358148}
{"user" : "venn", "name" : "venn", "pass" : 1561355361271}
{"user" : "venn", "name" : "venn", "pass" : 1561355364276}
{"user" : "venn", "name" : "venn", "pass" : 1561355367279}
{"user" : "venn", "name" : "venn", "pass" : 1561355370283}

run
flink : key : msgKey1561355358180, value {"user" : "venn", "name" : "venn", "pass" : 1561355358148}map
flink : key : msgKey1561355361271, value {"user" : "venn", "name" : "venn", "pass" : 1561355361271}map
flink : key : msgKey1561355364276, value {"user" : "venn", "name" : "venn", "pass" : 1561355364276}map
flink : key : msgKey1561355367279, value {"user" : "venn", "name" : "venn", "pass" : 1561355367279}map
flink : key : msgKey1561355370283, value {"user" : "venn", "name" : "venn", "pass" : 1561355370283}map
flink : key : msgKey1561355373289, value {"user" : "venn", "name" : "venn", "pass" : 1561355373289}map
flink : key : msgKey1561355376293, value {"user" : "venn", "name" : "venn", "pass" : 1561355376293}map

receive -> key : key1561355430411, value : key : msgKey1561355430356, value {"user" : "venn", "name" : "venn", "pass" : 1561355430356}map
receive -> key : key1561355433427, value : key : msgKey1561355433359, value {"user" : "venn", "name" : "venn", "pass" : 1561355433359}map
receive -> key : key1561355436441, value : key : msgKey1561355436364, value {"user" : "venn", "name" : "venn", "pass" : 1561355436364}map
receive -> key : key1561355439456, value : key : msgKey1561355439367, value {"user" : "venn", "name" : "venn", "pass" : 1561355439367}map
receive -> key : key1561355442473, value : key : msgKey1561355442370, value {"user" : "venn", "name" : "venn", "pass" : 1561355442370}map
receive -> key : key1561355445391, value : key : msgKey1561355445374, value {"user" : "venn", "name" : "venn", "pass" : 1561355445374}map