我想讀一個消息使用磚從一個支流卡夫卡的話題但是我繼續下麵的錯誤。你能讓我知道如果我遺漏了什麼東西?
代碼
選擇= {
“kafka.sasl.jaas.config”
:
“org.apache.kafka.common.security.plain.PlainLoginModule需要用戶名= " "密碼= " ";“
,
“kafka.sasl.mechanism”
:
“普通”
,
“kafka.security.protocol”
:
“SASL_SSL”
,
“kafka.bootstrap.servers”
:
“代理:9093”
,
“訂閱”
:
“sample_topic”
,
“kafka.group.id”
:
”“group_id
,
“startingOffsets”
:
“最早”
}
df =(火花
.readStream
。
格式
(
“卡夫卡”
)
.options(* *選項)
.load ())
json_df = df.selectExpr (
“鑄(值為字符串)作為json”
)
查詢= json_df.writeStream。
格式
(
“控製台”
).start ()
query.awaitTermination ()
錯誤
org.apache.spark.sql.kafka010.KafkaOffsetReaderConsumer.fetchEarliestOffsets (KafkaOffsetReaderConsumer.scala: 334) org.apache.spark.sql.kafka010.KafkaMicroBatchStream。anonfun getOrCreateInitialPartitionOffsets美元1美元(KafkaMicroBatchStream.scala: 302) scala.Option.getOrElse (Option.scala: 189) org.apache.spark.sql.kafka010.KafkaMicroBatchStream.getOrCreateInitialPartitionOffsets (KafkaMicroBatchStream.scala: 299) org.apache.spark.sql.kafka010.KafkaMicroBatchStream.initialOffset (KafkaMicroBatchStream.scala: 132) org.apache.spark.sql.execution.streaming.MicroBatchExecution。anonfun getStartOffset美元2美元(MicroBatchExecution.scala: 566) scala.Option.getOrElse (Option.scala: 189) org.apache.spark.sql.execution.streaming.MicroBatchExecution.getStartOffset (MicroBatchExecution.scala: 566)
