淘先锋技术网

首页 1 2 3 4 5 6 7

代码很简单,如下:

package com.imooc.flink.course08
import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.CheckpointingMode
object KafkaConnectorConsumerApp {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val topic = "flinktest"
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "192.168.64.233:9092")
    properties.setProperty("group.id", "pk")
    val data: DataStream[String] = env.addSource(new FlinkKafkaConsumer011[String](topic, new SimpleStringSchema(), properties))
    data.print()
    env.execute("KafkaConnectorConsumerApp")
  }
}

具体原因是:properties.setProperty(“bootstrap.servers”, “192.168.64.233:9092”),这个地方设置bootstrap server的地址,不管是ip还是域名,都需要在本地的hosts文件中配置ip和域名的映射关系,因为它是按节点名称进行查找的