sparkstreaming在创建kafka消费者后挂起

hmtdttj4  于 2021-06-07  发布在  Kafka
关注(0)|答案(0)|浏览(302)

我试图得到一个非常简单的Kafka+sparkstreaming集成。
在Kafka那边，我克隆了这个存储库(https://github.com/confluentinc/cp-docker-images)做了一个docker组合让zookeeper和kafka运行。我创建了一个名为“foo”的主题并添加了消息。在这种情况下，Kafka在29092端口上运行。
在spark方面，我的build.sbt文件如下所示：

  1. name := "KafkaSpark"
  3. version := "0.1"
  5. scalaVersion := "2.11.12"
  7. val sparkVersion = "2.2.0"
  9. libraryDependencies ++= Seq(
  10.    "org.apache.spark" %% "spark-core" % sparkVersion,
  11.    "org.apache.spark" %% "spark-sql" % sparkVersion,
  12.    "org.apache.spark" %% "spark-streaming" % sparkVersion,
  13.    "org.apache.spark" %% "spark-streaming-kafka-0-10" % sparkVersion
  14. )

我能够从终端消费数据中运行以下代码段：

  1. import org.apache.spark._
  2. import org.apache.spark.streaming._
  4. object SparkTest {
  6.   def main(args: Array[String]): Unit = {
  7.     val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  8.     val ssc = new StreamingContext(conf, Seconds(3))
  10.     val lines = ssc.socketTextStream("localhost", 9999)
  12.     val words = lines.flatMap(_.split(" "))
  13.     val pairs = words.map(word => (word, 1))
  14.     val wordCounts = pairs.reduceByKey(_ + _)
  16.     // Print the first ten elements of each RDD generated in this DStream to the console
  17.     wordCounts.print()
  18.     ssc.start()             // Start the computation
  19.     ssc.awaitTermination()  // Wait for the computation to terminate
  21.   }
  22. }

所以sparkstreaming在起作用。
现在，我从Kafka那里创建了以下内容：

  1. import org.apache.kafka.common.serialization.StringDeserializer
  2.  import org.apache.spark.sql.SparkSession
  3.  import org.apache.spark.sql.functions.count
  4.  import org.apache.spark.streaming.{Seconds, StreamingContext}
  5.  import org.apache.spark.streaming.kafka010._
  6.  import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
  7.  import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  8.  import org.apache.spark.sql.types.{StringType, StructType, TimestampType}
  10. object KafkaTest {
  12.   def main(args: Array[String]): Unit = {
  14.     val spark = SparkSession.builder
  15.       .master("local")
  16.       .appName("Spark Word Count")
  17.       .getOrCreate()
  19.     val ssc = new StreamingContext(spark.sparkContext, Seconds(3))
  21.     val kafkaParams = Map[String, Object](
  22.       "bootstrap.servers" -> "localhost:29092",
  23.       "key.deserializer" -> classOf[StringDeserializer],
  24.       "value.deserializer" -> classOf[StringDeserializer],
  25.       "group.id" -> "stream_group_id",
  26.       "auto.offset.reset" -> "latest",
  27.       "enable.auto.commit" -> (false: java.lang.Boolean)
  28.     )
  30.     val topics = Array("foo")
  31.     val stream = KafkaUtils.createDirectStream[String, String](
  32.       ssc,
  33.       PreferConsistent,
  34.       Subscribe[String, String](topics, kafkaParams)
  35.     )
  37.     stream.foreachRDD { (rdd, time) =>
  38.       val data = rdd.map(record => record.value)
  39.       data.foreach(println)
  40.       println(time)
  42.     }
  44.     ssc.start()             // Start the computation
  45.     ssc.awaitTermination()
  47.   }
  48. }

当它运行时，我在控制台中得到以下内容（我在intellij中运行它）。这个过程就挂在“订阅”主题后的最后一行。我尝试过创建一个不存在的主题，但我得到了相同的结果，也就是说，尽管没有主题存在，但它似乎不会抛出错误。如果我创建了一个不存在的代理，我会得到一个错误（线程“main”org.apache.kafka.common.kafkaexception中的异常：构造kafka使用者失败），所以当我使用正确的端口时，它一定是在查找代理。
对如何纠正这个问题有什么建议吗？
以下是日志文件：

  1. Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
  2. 17/11/23 05:29:42 INFO SparkContext: Running Spark version 2.2.0
  3. 17/11/23 05:29:42 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  4. 17/11/23 05:29:48 INFO SparkContext: Submitted application: Spark Word Count
  5. 17/11/23 05:29:48 INFO SecurityManager: Changing view acls to: jonathandick
  6. 17/11/23 05:29:48 INFO SecurityManager: Changing modify acls to: jonathandick
  7. 17/11/23 05:29:48 INFO SecurityManager: Changing view acls groups to: 
  8. 17/11/23 05:29:48 INFO SecurityManager: Changing modify acls groups to: 
  9. 17/11/23 05:29:48 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(jonathandick); groups with view permissions: Set(); users  with modify permissions: Set(jonathandick); groups with modify permissions: Set()
  10. 17/11/23 05:29:48 INFO Utils: Successfully started service 'sparkDriver' on port 59606.
  11. 17/11/23 05:29:48 DEBUG SparkEnv: Using serializer: class org.apache.spark.serializer.JavaSerializer
  12. 17/11/23 05:29:48 INFO SparkEnv: Registering MapOutputTracker
  13. 17/11/23 05:29:48 INFO SparkEnv: Registering BlockManagerMaster
  14. 17/11/23 05:29:48 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
  15. 17/11/23 05:29:48 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
  16. 17/11/23 05:29:48 INFO DiskBlockManager: Created local directory at /private/var/folders/w2/njgz3jnd097cdybxcvp9c2hw0000gn/T/blockmgr-3a3feb00-0fdb-4bc5-867d-808ac65d7c8f
  17. 17/11/23 05:29:48 INFO MemoryStore: MemoryStore started with capacity 2004.6 MB
  18. 17/11/23 05:29:48 INFO SparkEnv: Registering OutputCommitCoordinator
  19. 17/11/23 05:29:49 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
  20. 17/11/23 05:29:49 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.
  21. 17/11/23 05:29:49 WARN Utils: Service 'SparkUI' could not bind on port 4042. Attempting port 4043.
  22. 17/11/23 05:29:49 INFO Utils: Successfully started service 'SparkUI' on port 4043.
  23. 17/11/23 05:29:49 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://192.168.1.67:4043
  24. 17/11/23 05:29:49 INFO Executor: Starting executor ID driver on host localhost
  25. 17/11/23 05:29:49 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 59613.
  26. 17/11/23 05:29:49 INFO NettyBlockTransferService: Server created on 192.168.1.67:59613
  27. 17/11/23 05:29:49 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
  28. 17/11/23 05:29:49 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168.1.67, 59613, None)
  29. 17/11/23 05:29:49 INFO BlockManagerMasterEndpoint: Registering block manager 192.168.1.67:59613 with 2004.6 MB RAM, BlockManagerId(driver, 192.168.1.67, 59613, None)
  30. 17/11/23 05:29:49 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168.1.67, 59613, None)
  31. 17/11/23 05:29:49 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168.1.67, 59613, None)
  32. 17/11/23 05:29:49 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/Users/jonathandick/IdeaProjects/KafkaSpark/spark-warehouse/').
  33. 17/11/23 05:29:49 INFO SharedState: Warehouse path is 'file:/Users/jonathandick/IdeaProjects/KafkaSpark/spark-warehouse/'.
  34. 17/11/23 05:29:50 INFO StateStoreCoordinatorRef: Registered StateStoreCoordinator endpoint
  35. 17/11/23 05:29:50 WARN StreamingContext: spark.master should be set as local[n], n > 1 in local mode if you have receivers to get data, otherwise Spark jobs will not get resources to process the received data.
  36. 17/11/23 05:29:50 WARN KafkaUtils: overriding enable.auto.commit to false for executor
  37. 17/11/23 05:29:50 WARN KafkaUtils: overriding auto.offset.reset to none for executor
  38. 17/11/23 05:29:50 WARN KafkaUtils: overriding executor group.id to spark-executor-stream_group_id
  39. 17/11/23 05:29:50 WARN KafkaUtils: overriding receive.buffer.bytes to 65536 see KAFKA-3135
  40. 17/11/23 05:29:50 INFO DirectKafkaInputDStream: Slide time = 3000 ms
  41. 17/11/23 05:29:50 INFO DirectKafkaInputDStream: Storage level = Serialized 1x Replicated
  42. 17/11/23 05:29:50 INFO DirectKafkaInputDStream: Checkpoint interval = null
  43. 17/11/23 05:29:50 INFO DirectKafkaInputDStream: Remember interval = 3000 ms
  44. 17/11/23 05:29:50 INFO DirectKafkaInputDStream: Initialized and validated org.apache.spark.streaming.kafka010.DirectKafkaInputDStream@1a38eb73
  45. 17/11/23 05:29:50 INFO ForEachDStream: Slide time = 3000 ms
  46. 17/11/23 05:29:50 INFO ForEachDStream: Storage level = Serialized 1x Replicated
  47. 17/11/23 05:29:50 INFO ForEachDStream: Checkpoint interval = null
  48. 17/11/23 05:29:50 INFO ForEachDStream: Remember interval = 3000 ms
  49. 17/11/23 05:29:50 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream@1e801ce2
  50. 17/11/23 05:29:50 INFO ConsumerConfig: ConsumerConfig values: 
  51.     metric.reporters = []
  52.     metadata.max.age.ms = 300000
  53.     partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
  54.     reconnect.backoff.ms = 50
  55.     sasl.kerberos.ticket.renew.window.factor = 0.8
  56.     max.partition.fetch.bytes = 1048576
  57.     bootstrap.servers = [localhost:29092]
  58.     ssl.keystore.type = JKS
  59.     enable.auto.commit = false
  60.     sasl.mechanism = GSSAPI
  61.     interceptor.classes = null
  62.     exclude.internal.topics = true
  63.     ssl.truststore.password = null
  64.     client.id = 
  65.     ssl.endpoint.identification.algorithm = null
  66.     max.poll.records = 2147483647
  67.     check.crcs = true
  68.     request.timeout.ms = 40000
  69.     heartbeat.interval.ms = 3000
  70.     auto.commit.interval.ms = 5000
  71.     receive.buffer.bytes = 65536
  72.     ssl.truststore.type = JKS
  73.     ssl.truststore.location = null
  74.     ssl.keystore.password = null
  75.     fetch.min.bytes = 1
  76.     send.buffer.bytes = 131072
  77.     value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  78.     group.id = stream_group_id
  79.     retry.backoff.ms = 100
  80.     sasl.kerberos.kinit.cmd = /usr/bin/kinit
  81.     sasl.kerberos.service.name = null
  82.     sasl.kerberos.ticket.renew.jitter = 0.05
  83.     ssl.trustmanager.algorithm = PKIX
  84.     ssl.key.password = null
  85.     fetch.max.wait.ms = 500
  86.     sasl.kerberos.min.time.before.relogin = 60000
  87.     connections.max.idle.ms = 540000
  88.     session.timeout.ms = 30000
  89.     metrics.num.samples = 2
  90.     key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  91.     ssl.protocol = TLS
  92.     ssl.provider = null
  93.     ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
  94.     ssl.keystore.location = null
  95.     ssl.cipher.suites = null
  96.     security.protocol = PLAINTEXT
  97.     ssl.keymanager.algorithm = SunX509
  98.     metrics.sample.window.ms = 30000
  99.     auto.offset.reset = latest
  101. 17/11/23 05:29:50 DEBUG KafkaConsumer: Starting the Kafka consumer
  102. 17/11/23 05:29:50 INFO ConsumerConfig: ConsumerConfig values: 
  103.     metric.reporters = []
  104.     metadata.max.age.ms = 300000
  105.     partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
  106.     reconnect.backoff.ms = 50
  107.     sasl.kerberos.ticket.renew.window.factor = 0.8
  108.     max.partition.fetch.bytes = 1048576
  109.     bootstrap.servers = [localhost:29092]
  110.     ssl.keystore.type = JKS
  111.     enable.auto.commit = false
  112.     sasl.mechanism = GSSAPI
  113.     interceptor.classes = null
  114.     exclude.internal.topics = true
  115.     ssl.truststore.password = null
  116.     client.id = consumer-1
  117.     ssl.endpoint.identification.algorithm = null
  118.     max.poll.records = 2147483647
  119.     check.crcs = true
  120.     request.timeout.ms = 40000
  121.     heartbeat.interval.ms = 3000
  122.     auto.commit.interval.ms = 5000
  123.     receive.buffer.bytes = 65536
  124.     ssl.truststore.type = JKS
  125.     ssl.truststore.location = null
  126.     ssl.keystore.password = null
  127.     fetch.min.bytes = 1
  128.     send.buffer.bytes = 131072
  129.     value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  130.     group.id = stream_group_id
  131.     retry.backoff.ms = 100
  132.     sasl.kerberos.kinit.cmd = /usr/bin/kinit
  133.     sasl.kerberos.service.name = null
  134.     sasl.kerberos.ticket.renew.jitter = 0.05
  135.     ssl.trustmanager.algorithm = PKIX
  136.     ssl.key.password = null
  137.     fetch.max.wait.ms = 500
  138.     sasl.kerberos.min.time.before.relogin = 60000
  139.     connections.max.idle.ms = 540000
  140.     session.timeout.ms = 30000
  141.     metrics.num.samples = 2
  142.     key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
  143.     ssl.protocol = TLS
  144.     ssl.provider = null
  145.     ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
  146.     ssl.keystore.location = null
  147.     ssl.cipher.suites = null
  148.     security.protocol = PLAINTEXT
  149.     ssl.keymanager.algorithm = SunX509
  150.     metrics.sample.window.ms = 30000
  151.     auto.offset.reset = latest
  153. 17/11/23 05:29:50 INFO AppInfoParser: Kafka version : 0.10.0.1
  154. 17/11/23 05:29:50 INFO AppInfoParser: Kafka commitId : a7a17cdec9eaa6c5
  155. 17/11/23 05:29:50 DEBUG KafkaConsumer: Kafka consumer created
  156. 17/11/23 05:29:50 DEBUG KafkaConsumer: Subscribed to topic(s): foo
apache-kafkaapache-sparkspark-streaming

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com