Introduce new StreamingRunner

[thbar]

This PR introduces a new "Runner" implementation named StreamingRunner. The "Runner" is the part of Kiba which is responsible for carrying out the actual data processing.

Previously, a given "class transform" receiving one input row was only able to generate one output row, or no output row.

With this new runner, leveraging Ruby's Enumerator, each "class transform" can generate N output rows, by calling yield row, in addition to the row returned by process as usual.

While this can appear like something simple, it has a massive consequence, which is the ability to separate concerns when writing reusable Kiba components. This leads the way to writing more focused, generic & composable transforms when working with Kiba.

Let's pick an example.

Imagine you have created a Kiba source able to extract XML elements from a group of files on disk (with each file containing N elements).

It would typically look like this:

class MySource
  attr_reader :dir_pattern

  def initialize(dir_pattern:)
    @dir_pattern = dir_pattern
  end

  def each
    Dir[dir_pattern].sort.each do |file|
      doc = Nokogiri::XML(IO.binread(file))
      doc.search('/invoices/invoice').each do |item|
        yield(item)
      end
    end
  end
end

Such a class has 4 responsabilities:

• The directory iteration
• The XML parsing
• The XML node research
• The act of exploding each detected sub-node as an independent row

With Kiba 1 you can achieve some level of splitting here by using the decorator technique I outlined here, but this can only take you so far.

With the new Kiba runner, you can first rewrite the code above as 4 independent components (1 source & 3 transforms):

class DirectoryLister
  def initialize(dir_pattern:)

  def each
    Dir[dir_pattern].sort.each do |filename|
      yield(filename)
    end
  end
end

class XMLReader
  def process(filename)
    Nokogiri::XML(IO.binread(filename))
  end
end

class XMLSearcher
  def initialize(selector:)

  def process(doc)
    doc.search(selector)
  end
end

class EnumerableExploder
  def process(row)
    row.each do |item|
      yield(item)
    end
    nil # tell the pipeline to ignore the final value
  end
end   

Which can then be used:

source DirectoryLister, dir_pattern: '*.csv'
transform XMLReader
transform XMLSearcher, selector: '/invoices/invoice'
transform EnumerableExploder

While it can appear to be more complicated at first, each of these 4 components can now be mix-and-matched with other components in completely unrelated scenarios.

For instance:

• The DirectoryLister could be used to list anything (JSON files etc).
• The EnumerableExploder, similarly, could be used for pretty much anything.

This opens the door to provide more composable & more reusable components to Kiba users, or as part of Kiba Common or Kiba Pro.

How to enable the StreamingRunner

Using the new runner is opt-in. You'll have to do:

extend Kiba::DSLExtensions::Config

config :kiba, runner: Kiba::StreamingRunner

From there, you can write "yielding class transforms", like:

class MyTransform
  def process(row)
    yield {key: 1}
    yield {key: 2}
    {key: 3}
  end
end

Limitations

You can only yield rows from "class transforms". Calling yield from a "block transform" will generate an error.

Benchmark

Real-life benchmarks I made showed that the impact (on real-life examples) is negligible at this point.

A non-real-life benchmark used in 82bc8e5 shows that the new runner takes 5% to 30% more time. It's a non-real-life example because there are no real sources nor real destinations.

Notes on current implementation

This was initiated in a private repo, later published as #41, which has been reworked & finalized here.

[vfonic]

This is amazing feature! Thank you so much for taking the time to add it! ❤️

[thbar]

@vfonic thanks for your feedback! Much appreciated ^_^ 💙 💚