使用Kiba-ETL将表转换为集合哈希

Question

我正忙着通过ETL管道工作，但是对于这个特殊的问题，我需要获取一个数据表，并将每一列转换为一个集合，即一个唯一的数组。

我正努力思考如何在Kiba框架内完成这一任务。

以下是我所努力实现的目标的实质：

源

[
  { 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"],
}

这样的事情在Kiba是可行的吗？( b)在Kiba这样做更好吗？

任何帮助都将不胜感激。

更新-部分解决了。

我找到了一个部分的解决方案。这个转换器类将把一个行表转换成一个集合的散列，但是我一直在研究如何使用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

您的解决方案将工作得很好，事实上，在Kiba中使用这样的设计(主要是“普通老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"

通过检测输入数据集中的边界，甚至可以在目标中有多个行，从而产生相应的结果。

一旦可能的话，我会更新这个答案。

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.

现在，我轻松地转换了我的数据:)