如何在spark中读取压缩的avro文件(.gz)？

import org.apache.avro.generic.GenericRecord
import org.apache.avro.mapred.{AvroInputFormat, AvroWrapper}
import org.apache.hadoop.io.NullWritable

val spark = SparkSession.builder().appName("Read Compressed Avro File in Spark").getOrCreate()

val inputPath = "path/to/compressed_avro.gz"
spark.conf.set("spark.hadoop.mapred.input.dir.recursive", "true") // 对于目录递归搜索
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputdir", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputdir.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputdir.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs", inputPath)
spark.conf.set("spark.hadoop.mapred.input.dir", inputPath)
spark.conf.set("spark.hadoop.mapred.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dir.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.list.status", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.listing.prune", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.num-threads", "5")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.threads.sleep-ms", "1000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.interval-ms", "5000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.offset-interval-ms", "5000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.threads", "5")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.file-cache-size", "5000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.use-partial-listing", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.time-interval", "5000")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.maxsize", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.node", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.rack", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.non-recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.num-threads", "5")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.node.calculator", "org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat$OneBlockPerSplit")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.rack.calculator", "org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat$OneBlockPerSplit")
spark.conf.set("spark.hadoop.mapred.input.format.class", classOf[AvroInputFormat[GenericRecord]].getName)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputformat", classOf[AvroInputFormat[GenericRecord]].getName)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputformat", "org.apache.hadoop.mapreduce.lib.input.AvroKeyInputFormat")
spark.conf.set("spark.hadoop.mapred.input.format.class", "org.apache.hadoop.mapreduce.lib.input.AvroKeyInputFormat")
spark.conf.set("spark.hadoop.mapred.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapred.input.pathFilter.class", "org.apache.hadoop.fs.PathFilter")
spark.conf.set("spark.hadoop.mapred.input.pathFilter.class", "org.apache.hadoop.mapred.SimpleRegexPathFilter")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.num-threads", "5")
spark.conf.set("spark.hadoop.mapred.input.dir.listing.interval-ms", "5000")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.threads.sleep-ms", "1000")
spark.conf.set("spark.hadoop.mapred.input.dir.listing.prune.file-cache-size", "5000")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.list-status.threads.sleep-ms", "1000")
spark.conf.set("spark.hadoop.mapred.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.inputdir", inputPath)
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir", inputPath)
spark.conf.set("spark.hadoop.mapred.input.dir", inputPath)
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.interval-ms", "5000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.offset-interval-ms", "5000")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.maxsize", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.use-partial-listing", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.time-interval", "5000")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.node", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.node.calculator", "org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat$OneBlockPerSplit")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.rack", "134217728")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize.per.rack.calculator", "org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat$OneBlockPerSplit")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.non-recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.non-recursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.threads", "5")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.recursive", "true")
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.input.dirs.nonrecursive", "true")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.file-cache-size", "5000")
spark.conf.set("spark.hadoop.mapred.input.dirs.listing.prune.threads", "5")
spark.conf.set("spark.hadoop.mapred.input.format.class", classOf[AvroInputFormat[GenericRecord]].getName)

val avroRDD = spark.sparkContext.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](inputPath)
val records = avroRDD.map(record => record._1.get())

records.foreach(println)