当我消费和接收Kafka的信息时,我遇到了一些奇怪的错误。我现在运行的是2.3.0,我知道这在其他版本之前是可行的。
val event = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", <server list>)
.option("subscribe", <topic>)
.load()
val filesink_query = outputdf.writeStream
.partitionBy(<some column>)
.format("parquet")
.option("path", <some path in EMRFS>)
.option("checkpointLocation", "/tmp/ingestcheckpoint")
.trigger(Trigger.ProcessingTime(10.seconds))
.outputMode(OutputMode.Append)
.start
java.lang.IllegalStateException: /tmp/outputagent/_spark_metadata/0 doesn't exist when compacting batch 9 (compactInterval: 10)
我很困惑,这是最新版本spark中的错误吗?
1条答案
按热度按时间oogrdqng1#
这个问题似乎和在s3a上使用s3n和只在hdfs上有检查点而不是s3有关。这是非常烦人的sine我想避免硬编码dns或ips在我的代码。