Ruby: leitura, an√°lise e encaminhamento de grandes arquivos JSON em pequenos blocos (ou seja, streaming)

Passei alguns dias tentando descobrir isso e pensei em compartilhar uma dica para evitar o aborrecimento de outra pessoa.

O problema : eu tenho uma API Ruby e tenho que ler arquivos JSON enormes de fontes de terceiros, analisá-los e fazer coisas com os dados antes de encaminhá-los para outro lugar. Isso pode resultar claramente em um pesadelo de memória se tudo tiver que ser feito na memória.

Parece haver bastante conhecimento da comunidade sobre como transmitir dados em Ruby do servidor para o cliente (por exemplo, aqui ), mas n√£o tanto sobre como fazer o mesmo ao ler um fluxo JSON.

Embora você acabe passando os dados por meio de Enumeradores Ruby em ambos os casos, há uma diferença fundamental:

  • Ao enviar dados, sabemos que o JSON produzido √© v√°lido e queremos apenas grav√°-lo em um stream sem fazer mais nada;
  • Ao ler dados, se quisermos consumi-los sem manter todo o documento na mem√≥ria, precisamos entend√™-los √† medida que chegam, sem saber como o documento ir√° evoluir.

Eu encontrei um artigo muito bom sobre como consumir fluxos XML, aqui . Ele ainda vem com uma gema chamada Piperator, que permite que você encadeie etapas em seu pipeline de uma forma limpa e legível.

Com um pouco de ajuda do gem, tentei implementar o mesmo em JSON usando o analisador JSON Oj , que li supera todos os outros por aí.

Aqui está seu exemplo, onde os itens principais a serem verificados são runos yield_chunkmétodos e:

require 'oj'
require
'piperator'

class JSONStreamParser < ::Oj::ScHandler
def initialize
@data_stream_writer = Oj::StringWriter.new
@running = false
end

def run(enumerable_data_source, &block)
if !@running
@running = true
@yielder = block

# This turns your enumerator into an IO class, very handy
# as Oj's sc_parse method wants an IO object.
io
= Piperator::IO.new(enumerable_data_source)
Oj.sc_parse(self, io)
end
end

def hash_key(key)
update_current_path
(:hash_key, key)
@data_stream_writer.push_key(key)
@last_key = key
end

def hash_start
@data_stream_writer.push_object(@last_key)
@last_key = nil
end

def hash_set(h, key, value)
@data_stream_writer.push_value(value, key)
end

def hash_end
@data_stream_writer.pop
yield_if_condition

end

def array_start
@data_stream_writer.push_array(@last_key)
@last_key = nil
end

def array_append(a, value)
@data_stream_writer.push_value(value) unless !value && @array_ended
@array_ended = false
end

def array_end
@data_stream_writer.pop
@array_ended = true
yield_if_condition

end

def add_value(value)
@data_stream_writer.push_value(value, @last_key)
@last_key = nil
end

def error(message, line, column)
p
"ERROR: #{message}"
end

private

def yield_if_condition
# if whatever_logic
# @data_stream_writer.pop_all
# yield_chunk
# @data_stream_writer = Oj::StringWriter.new
# [ further logic depending on data structure ]
# end
end

def yield_chunk
@yielder.call @data_stream_writer.to_s
end
end

http_fetch
= Enumerator.new do |yielder|
url
= "https://raw.githubusercontent.com/zemirco/sf-city-lots-json/master/citylots.json"
request
= Typhoeus::Request.new(url)
request
.on_body do |chunk|
yielder
<< chunk
end
request
.run
end

json_parse
= Enumerator.new do |yielder|
parser
= JSONStreamParser.new
parser
.run(http_fetch) do |parsed_chunk|
yielder
<< parsed_chunk
end
end

json_parse
.map{ |c| puts c }