赞
踩
简单的flink接入kafka结果报错:
---------------------------
————————————————————————————————————————
//create env
val env = StreamExecutionEnvironment.getExecutionEnvironment
//set parallelism
env.setParallelism(1)
//set process time
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
//for kafka connection
val kafkaProps = new Properties()
//kafka's attribute
kafkaProps.setProperty("bootstrap.servers","192.168.***.***:9092")
//set the consumer's group
kafkaProps.setProperty("group.id","group1")
//create the consumer
val kafkaSource = new FlinkKafkaConsumer[String]("stockPrice", new SimpleStringSchema, kafkaProps)
//set offset
kafkaSource.setStartFromEarliest()
//auto commit offset
kafkaSource.setCommitOffsetsOnCheckpoints(true)
//band data source
val ds = env.addSource(kafkaSource)
ds.print()
env.execute()
zz的原因:
IP写错了
我用KafkaTools工具连接kafka 连不上,报
timeout expired while fetching topic metadata
这个错误
,然后四处百度,发现了问题,
解决办法是修改kafka的kafka_2.11-2.1.0/config/路径下的server.properties文件
advertised.listeners=PLAINTEXT://zjj101:9092
zjj101是当前服务器的ip,然后保存配置,重启kafka, 即可解决这个问题.
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。