Nano Hash - криптовалюты, майнинг, программирование

Spark не читает все записи из двоичного файла

Я пытаюсь читать файлы Avro с S3 и, как показано в этом документация по искре Я могу ее нормально прочитать. Мои файлы, как показано ниже, эти файлы состоят из 5000 записей каждый.

s3a://bucket/part-0.avro
s3a://bucket/part-1.avro
s3a://bucket/part-2.avro

val byteRDD: RDD[Array[Byte]] = sc.binaryFiles(s"$s3URL/*.avro").map{ case(file, pds) => {
  val dis = pds.open()
  val len = dis.available()
  val buf = Array.ofDim[Byte](len)
  pds.open().readFully(buf)
  buf
}}

import org.apache.avro.io.DecoderFactory
val deserialisedAvroRDD = byteRDD.map(record => {

  import org.apache.avro.Schema
  val schema = new Schema.Parser().parse(schemaJson)
  val datumReader = new GenericDatumReader[GenericRecord](schema)

  val decoder = DecoderFactory.get.binaryDecoder(record, null)
  var datum: GenericRecord = null
  while (!decoder.isEnd()) {
    datum = datumReader.read(datum, decoder)
  }
  datum
}
)

deserialisedAvroRDD.count() ---> 3

Я десериализую сообщения binaryAvro для создания GenericRecords, и я ожидал, что десериализованный RDD будет иметь 15 000 записей, поскольку каждый файл .avro имеет 5 000 записей, однако после десериализации я получаю только 3 записи. Может ли кто-нибудь помочь в поиске проблемы с моим кодом? Как я могу сериализовать одну запись за раз.

apache-spark deserialization avro spark-avro binaryfiles
12.05.2020

  • Отвечает ли это на ваш вопрос? Чтение файла Avro в Spark 13.05.2020
  • Это Binary Avros, т.е. Array[Bytes] 13.05.2020
  • Проблема скорее всего в byteRDD чтении. Он не знает, когда запись начинается и останавливается. Есть ли причина разделить операции на 2 разных этапа? Почему бы не использовать binaryDecoder для чтения pds? 13.05.2020
  • Почему бы не использовать binaryDecoder для чтения pds? -› Можете ли вы указать пример для этого? 13.05.2020

Ответы:


1

Это должно работать

val recRDD: RDD[GenericRecord] = sc.binaryFiles(s"$s3URL/*.avro").flatMap {
  case (file, pds) => {
    val schema =  new Schema.Parser().parse(schemaJson)
    val datumReader = new GenericDatumReader[GenericRecord](schema)

    val decoder = DecoderFactory.get.binaryDecoder(pds.toArray(), null)
    var datum: GenericRecord = null
    val out = ArrayBuffer[GenericRecord]()
    while (!decoder.isEnd()) {
      out += datumReader.read(datum, decoder)
    }
    out
  }
}
13.05.2020
Новые материалы
Кластеризация: более глубокий взгляд
Кластеризация — это метод обучения без учителя, в котором мы пытаемся найти группы в наборе данных на основе некоторых известных или неизвестных свойств, которые могут существовать. Независимо от..
Как написать эффективное резюме
Предложения по дизайну и макету, чтобы представить себя профессионально Вам не позвонили на собеседование после того, как вы несколько раз подали заявку на работу своей мечты? У вас может..
Частный метод Python: улучшение инкапсуляции и безопасности
Введение Python — универсальный и мощный язык программирования, известный своей простотой и удобством использования. Одной из ключевых особенностей, отличающих Python от других языков, является..
Как я автоматизирую тестирование с помощью Jest
Шутка для победы, когда дело касается автоматизации тестирования Одной очень важной частью разработки программного обеспечения является автоматизация тестирования, поскольку она создает..
Работа с векторными символическими архитектурами, часть 4 (искусственный интеллект)
Hyperseed: неконтролируемое обучение с векторными символическими архитектурами (arXiv) Автор: Евгений Осипов , Сачин Кахавала , Диланта Хапутантри , Тимал Кемпития , Дасвин Де Сильва ,..
Понимание расстояния Вассерштейна: мощная метрика в машинном обучении
В обширной области машинного обучения часто возникает необходимость сравнивать и измерять различия между распределениями вероятностей. Традиционные метрики расстояния, такие как евклидово..
Обеспечение масштабируемости LLM: облачный анализ с помощью AWS Fargate и Copilot
В динамичной области искусственного интеллекта все большее распространение получают модели больших языков (LLM). Они жизненно важны для различных приложений, таких как интеллектуальные..