Я пытаюсь прочитать файлы из нескольких блоков s3.
Первоначально ведра были бы в разных регионах, но, похоже, это невозможно.
Итак, теперь я скопировал другой сегмент в ту же область, что и первый сегмент для чтения, из той же области, из которой я выполняю задание spark.
Настройка SparkSession:
val sparkConf = new SparkConf()
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Event]))
SparkSession.builder
.appName("Merge application")
.config(sparkConf)
.getOrCreate()
Функция, которая вызывается с использованием SQLContext из create SparkSession:
private def parseEvents(bucketPath: String, service: String)(
implicit sqlContext: SQLContext
): Try[RDD[Event]] =
Try(
sqlContext.read
.option("codec", "org.apache.hadoop.io.compress.GzipCodec")
.json(bucketPath)
.toJSON
.rdd
.map(buildEvent(_, bucketPath, service).get)
)
Основной поток:
for {
bucketOnePath <- buildBucketPath(config.bucketOne.name)
_ <- log(s"Reading events from $bucketOnePath")
bucketOneEvents: RDD[Event] <- parseEvents(bucketOnePath, config.service)
_ <- log(s"Enriching events from $bucketOnePath with originating region data")
bucketOneEventsWithRegion: RDD[Event] <- enrichEventsWithRegion(
bucketOneEvents,
config.bucketOne.region
)
bucketTwoPath <- buildBucketPath(config.bucketTwo.name)
_ <- log(s"Reading events from $bucketTwoPath")
bucketTwoEvents: RDD[Event] <- parseEvents(config.bucketTwo.name, config.service)
_ <- log(s"Enriching events from $bucketTwoPath with originating region data")
bucketTwoEventsWithRegion: RDD[Event] <- enrichEventsWithRegion(
bucketTwoEvents,
config.bucketTwo.region
)
_ <- log("Merging events")
mergedEvents: RDD[Event] <- merge(bucketOneEventsWithRegion, bucketTwoEventsWithRegion)
if mergedEvents.isEmpty() == false
_ <- log("Grouping merged events by partition key")
mergedEventsByPartitionKey: RDD[(EventsPartitionKey, Iterable[Event])] <- eventsByPartitionKey(
mergedEvents
)
_ <- log(s"Storing merged events to ${config.outputBucket.name}")
_ <- store(config.outputBucket.name, config.service, mergedEventsByPartitionKey)
} yield ()
Ошибка, которую я получаю в журналах (реальные имена блоков были изменены, но настоящие имена существуют):
19/04/09 13:10:20 INFO SparkContext: Created broadcast 4 from rdd at MergeApp.scala:141
19/04/09 13:10:21 INFO FileSourceScanExec: Planning scan with bin packing, max size: 134217728 bytes, open cost is considered as scanning 4194304 bytes.
org.apache.spark.sql.AnalysisException: Path does not exist: hdfs:someBucket2
И мои журналы stdout показывают, как далеко проходит основной код до сбоя:
Reading events from s3://someBucket/*/*/*/*/*.gz
Enriching events from s3://someBucket/*/*/*/*/*.gz with originating region data
Reading events from s3://someBucket2/*/*/*/*/*.gz
Merge failed: Path does not exist: hdfs://someBucket2
Странно, первое чтение всегда работает независимо от того, какое ведро я выберу.
Но второе чтение всегда терпит неудачу, независимо от ведра.
Это говорит мне, что с корзинами нет ничего плохого, но есть некоторая странность при работе с несколькими корзинами s3.
Я вижу только потоки при чтении нескольких файлов из одного сегмента s3, но не нескольких файлов из нескольких блоков s3.
Есть идеи?