Преобразование таблицы в хэш наборов с использованием Kiba-ETL

Question

Я занят работой через конвейер ETL, но для этой конкретной проблемы мне нужно взять таблицу данных и превратить каждый столбец в набор, то есть в уникальный массив.

Я изо всех сил пытаюсь обернуть голову, как бы я достиг этого в рамках Киба.

Вот суть того, чего я пытаюсь достичь:

Источник:

[
  { dairy: "Milk",   protein: "Steak",   carb: "Potatoes" },
  { dairy: "Milk",   protein: "Eggs",    carb: "Potatoes" },
  { dairy: "Cheese", protein: "Steak",   carb: "Potatoes" },
  { dairy: "Cream",  protein: "Chicken", carb: "Potatoes" },
  { dairy: "Milk",   protein: "Chicken", carb: "Pasta" },
]

Место назначения

{
  dairy:   ["Milk", "Cheese", "Cream"],
  protein: ["Steak", "Eggs", "Chicken"],
  carb:    ["Potatoes", "Pasta"],
}

Является ли что-то подобное а) выполнимым в Кибе и б) даже целесообразным в Кибе?

Любая помощь будет принята с благодарностью.

Обновление - частично решено.

Я нашел частичное решение. Этот класс преобразователя преобразует таблицу строк в хэш наборов, но я застрял на том, как получить эти данные, используя назначение ETL. Я подозреваю, что я использую Kiba таким образом, что он не предназначен для использования.

class ColumnSetTransformer
  def initialize
    @col_set = Hash.new(Set.new)
  end

  def process(row)
    row.each do |col, col_val|
      @col_set[col] = @col_set[col] + [col_val]
    end

    @col_set
  end
end 

Answer 1

Ваше решение будет работать просто отлично, и причина того, что такой дизайн в Кибе (в основном "простые старые объекты Ruby"), заключается в том, чтобы облегчить вызов компонентов самостоятельно, если вам это нужно! (это очень полезно для тестирования!).

Тем не менее, здесь есть несколько дополнительных возможностей.

То, что вы делаете, является формой агрегации, которая может быть реализована различными способами.

Буферизация назначения

Здесь буфер будет одной строкой, на самом деле. Используйте код, такой как:

class MyBufferingDestination
  attr_reader :single_output_row

  def initialize(config:)
    @single_output_row = []
  end

  def write(row)
    row.each do |col, col_val|
      single_output_row[col] += [col_val]
    end
  end

  def close # will be called by Kiba at the end of the run
    # here you'd write your output
  end
end

Использование переменной экземпляра для агрегирования + блок post_process

pre_process do
  @output_row = {}
end

transform do |row|
  row.each do |col, col_val|
    @output_row = # SNIP
  end      
  row
end

post_process do
  # convert @output_row to something
  # you can invoke a destination manually, or do something else
end

Вскоре возможно: с использованием преобразования буферизации

Как описано здесь, скоро станет возможным создание буферных преобразований, чтобы лучше отделить механизм агрегирования от самого места назначения.

Это пойдет так:

class MyAggregatingTransform
  def process(row)
    @aggregate += xxx
    nil # remove the row from the pipeline
  end

  def close
    # not yet possible, but soon
    yield @aggregate
  end
end

Это будет лучший дизайн, потому что тогда вы сможете повторно использовать существующие места назначения, не изменяя их для поддержки буферизации, чтобы они стали более общими и пригодными для повторного использования:

transform MyAggregatingTransform

destination MyJSONDestination, file: "some.json"

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

Я обновлю SO ответ, как только это станет возможным.

Answer 2

Итак, использование Kiba в контексте работы не похоже на то, как этот инструмент предназначался для использования. Я хотел использовать Kiba, потому что я уже реализовал много кода E, T и L для этого проекта, и его повторное использование было бы огромным.

Итак, если у меня есть код для повторного использования, но я не могу использовать его в рамках Kiba, я могу просто назвать его, как если бы он был нормальным кодом. Это все благодаря превосходно простому дизайну Thibaut!

Вот как я решил проблему:

source  = CSVOrXLSXSource.new("data.xlsx", document_config: { some: :settings })
xformer = ColumnSetTransformer.new

source.each do |row|
  xformer.process(row)
end

p xformer.col_set # col_set must be attr_reader on this class.

И теперь мои данные легко преобразуются:)