资讯

展开

Kafka-Spark-(Producer)相关API(将数据存入指定分区)

作者:快盘下载 人气:
object MyProducer extends App {
    //配置
    val properties = new Properties()
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,;master:9092,slave1:9092,slave2:9092;)
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer].getName)
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer].getName)
    //创建生产者
    val kafkaProducer = new KafkaProducer[String,String](properties)
    for (i <- 0 to 100) {
      kafkaProducer.send(new ProducerRecord(;reback;, 3,;;, ;; ; i),new Callback {
        //将值存入指定的分区里面 topic = ;reback;,partition = 3,key = ;;,value = ;; ; i
        override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {
          if (e == null)
            println(;主题:; ; recordMetadata.topic() ; ;分区:; ; recordMetadata.partition() ; ;值长度:;
              ; recordMetadata.serializedValueSize())//serializedValueSize;返回序列化后值得长度
          else e.printStackTrace()
        }
      })
    }
    kafkaProducer.close();
}

Kafka-Spark-(Producer)相关API(将数据存入指定分区)

加载全部内容

相关教程
猜你喜欢
用户评论
快盘暂不提供评论功能!