代码很简单,如下:

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和域名的映射关系,因为它是按节点名称进行查找的

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐