scala - Spark消费者不阅读Kafka生产者消息Scala
问题描述
我正在尝试创建连接到 Spark 消费者的 Kafka 生产者。生产者工作正常,但是 Spark 中的消费者由于某种原因没有从主题中读取数据。我在 docker-compose 中使用 spotify/kafka 图像运行 kafka。
这是我的消费者:
object SparkConsumer {
def main(args: Array[String]) {
val spark = SparkSession
.builder()
.appName("KafkaSparkStreaming")
.master("local[*]")
.getOrCreate()
val ssc = new StreamingContext(spark.sparkContext, Seconds(3))
val topic1 = "topic1"
def kafkaParams = Map[String, Object](
"bootstrap.servers" -> "localhost:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "group1",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val lines = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Set(topic1), kafkaParams)
)
lines.print()
}
Kafka Producer 如下所示:
object KafkaProducer {
def main(args: Array[String]) {
val events = 10
val topic = "topic1"
val brokers = "localhost:9092"
val random = new Random()
val props = new Properties()
props.put("bootstrap.servers", brokers)
props.put("client.id", "KafkaProducerExample")
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
val producer = new KafkaProducer[String, String](props)
val t = System.currentTimeMillis()
for (nEvents <- Range(0, events)) {
val key = null
val values = "2017-11-07 04:06:03"
val data = new ProducerRecord[String, String](topic, key, values)
producer.send(data)
System.out.println("sent : " + data.value())
}
System.out.println("sent per second: " + events * 1000 / (System.currentTimeMillis() - t))
producer.close()
}
}
更新:
我使用 Kafka 的 docker-compose 文件:
version: '3.3'
services:
kafka:
image: spotify/kafka
ports:
- "9092:9092"
解决方案
这是将 Kafka 与 Docker 结合使用的常见问题。首先,您应该检查 Zookeeper 中针对您的主题的配置。您可以在 Kafka 容器中使用 Zookeeper 脚本。可能在创建主题时 ADVERTISED_HOST 是您的服务的名称。因此,当消费者尝试连接到代理时,这将返回“kafka”作为代理位置。因为您在 docker 网络之外运行消费者,所以您的消费者永远不会连接到代理进行消费。尝试使用 ADVERTISED_HOST=localhost 为您的 kafka 容器设置环境。
推荐阅读
- javascript - Node.js 可读流事件“关闭”和“结束”之间的确切区别是什么?
- xsd - 在 XSD 中创建动态属性
- c++ - “Clang 中的 C++ 支持”页面中的 SVN 是什么意思?
- asp.net-mvc-viewmodel - MVC 视图中的 if else 条件
- python - 抓取不断更新的网页
- image - 有没有办法直接从字节创建光栅图像以用于画布(不显示)显示?
- google-cloud-platform - 循环遍历许多文件时如何增加gcsfuse吞吐量?
- c# - AAD 身份验证和 cookie 名称?
- r - r 在绘制日内数据时如何从 x 轴中删除工作日的非工作时间(以及周末和节假日)
- python - scikit k-means:查找属于特定质心生成集群的数据点