我有一个Spark结构化流应用程序,必须立即读取12个Kafka主题(不同的架构,Avro格式),反序列化数据并存储在HDFS中。当我使用我的代码从单个主题中读取内容时,它可以正常工作且没有错误,但同时运行多个查询时,出现以下错误
java.lang.IllegalStateException: Race while writing batch 0
我的代码如下:
def main(args: Array[String]): Unit = {
val kafkaProps = Util.loadProperties(kafkaConfigFile).asScala
val topic_list = ("topic1", "topic2", "topic3", "topic4")
topic_list.foreach(x => {
kafkaProps.update("subscribe", x)
val source= Source.fromInputStream(Util.getInputStream("/schema/topics/" + x)).getLines.mkString
val schemaParser = new Schema.Parser
val schema = schemaParser.parse(source)
val sqlTypeSchema = SchemaConverters.toSqlType(schema).dataType.asInstanceOf[StructType]
val kafkaStreamData = spark
.readStream
.format("kafka")
.options(kafkaProps)
.load()
val udfDeserialize = udf(deserialize(source), DataTypes.createStructType(sqlTypeSchema.fields))
val transformedDeserializedData = kafkaStreamData.select("value").as(Encoders.BINARY)
.withColumn("rows", udfDeserialize(col("value")))
.select("rows.*")
val query = transformedDeserializedData
.writeStream
.trigger(Trigger.ProcessingTime("5 seconds"))
.outputMode("append")
.format("parquet")
.option("path", "/output/topics/" + x)
.option("checkpointLocation", checkpointLocation + "//" + x)
.start()
})
spark.streams.awaitAnyTermination()
}
您可以使用KAFKA Connect(来自Confluent),NIFI,StreamSets等,因为您的用例似乎适合“向HDFS转储/持久化”。就是说,您需要安装这些工具。您声明的小文件问题不是问题,就这样。
从Apache Kafka 0.9或更高版本,您可以为KAFKA使用Kafka Connect API-> HDFS Sink(各种受支持的HDFS格式)。尽管您需要一个KAFKA Connect群集,但是无论如何它都基于您现有的群集,所以没什么大不了的。但是有人需要维护。
一些在您途中与您联系的链接: