Потоковое декодирование JSON с использованием предпочтительно Circe и Akka Streams

Мой вариант использования похож на эту запись, я хочу прочитать внутреннюю, огромный массив (несколько гигабайт в виде текста) из объекта JSON, например:

{ "a": "...",   // root level fields to be read, separately
  ...
  "bs": [       // the huge array, most of the payload (can be multiple GB's)
    {...},
    ...
  ]
}

Вход доступен как Source[ByteString,_] (поток Akka), и я использую Circe для декодирования JSON в другом месте.

Я вижу две проблемы:

  1. Чтение массива bs в потоковом режиме (получение Source[B,_] за его потребление).

  2. Разделение исходного потока на два, чтобы я мог читать и анализировать поля корневого уровня до начала массива.

У вас есть указатели на решение такого варианта использования? Я проверил akka-stream-json и circe-iteratee.

akka-stream-json выглядит как вещь, но в не очень хорошем состоянии. circe-iteratee похоже, не имеет интеграции с Akka Streams.


json akka-stream circe
person akauppi    schedule 19.07.2018    source источник

Ответы (2)


arrow_upward
1
arrow_downward

У Jawn есть асинхронный парсер: https://github.com/non/jawn/blob/master/parser/src/main/scala/jawn/AsyncParser.scala

Но написать эффективный асинхронный парсер для JSON сложно из-за его последовательного происхождения.

Если вы можете переключиться на синхронный синтаксический анализ, вы можете использовать jsoniter-scala-core и написать простой пользовательский кодек, который пропустит все ненужные пары ключ / значение, кроме «bs», а затем проанализирует требуемые данные невероятно быстро без хранения или содержимого массива в объем памяти.

person Andriy Plokhotnyuk    schedule 23.07.2018
comment
akka-stream-json строится поверх Jawn. Проблема, с которой я столкнулся, заключается в чтении ›файла JSON размером 5 ГБ, где основная часть представляет собой единый массив. - person akauppi; 23.07.2018

arrow_upward
0
arrow_downward

Я вижу, что для потокового декодирования JSON нужна совершенно новая библиотека.

Что-то типа:

case class A(a: Int, bs: Source[B,_])

val src: Source[ByteString,_] = ???
src.as[A]

Мое временное решение - "массировать" JSON с помощью jq и sed, так что каждый B находится на своей собственной строке. Таким образом, я могу использовать исходный код по строкам и декодировать каждый B отдельно.

Вот сценарий Bash (без гарантий):

#!/bin/bash

arrKey=$1
input=$2

head -n 1 $input | sed s/.$//
jq -M -c ".$arrKey|.[]" $input | sed s/$/,/
echo "]}"

Он действительно полагается на определенные вещи, например вопрос, не являющийся массивом, всегда находится в первой строке (а они и есть).

person akauppi    schedule 20.07.2018