结构化流媒体：同时阅读多个Kafka主题

Question

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()  
 }

Answer 1

另类。您可以使用KAFKA Connect（来自Confluent)、NIFI、StreamSets等，因为您的用例似乎适合“dump/persisted to HDFS”。也就是说，您需要安装这些工具。您所说的小文件问题不是一个问题，所以就这样吧。

在Apache Kafka0.9或更高版本中，您可以使用Kafka Connect API for Kafka-->HDFS接收器（各种支持的HDFS格式）。不过，您需要一个KAFKA Connect集群，但无论如何，这都是基于现有集群的，所以没什么大不了的。但需要有人维护。

一些让你上路的链接：

null