使用 Kiba 时是否有明显的方法来减少行数？答案

【问题标题】：Is there an obvious way to reduce rows when using Kiba?使用 Kiba 时是否有明显的方法来减少行数？
【发布时间】：2020-03-31 10:39:23
【问题描述】：

首先 - Thibaut，感谢 Kiba。它与“企业”级 ETL 工具相得益彰，从未让我失望。

我正忙于构建一个 ETL 管道，该管道包含许多行，并将它们缩减为单个摘要行。我觉得这应该是一件简单的事情，但我对如何解决这个问题有点困惑。

我们有许多来自语音交换机的 CDR，需要根据一些简单的标准将它们压缩成一些摘要记录。所以，问题是；我有数千条来自Source 的记录，需要根据一些减少标准将它们转换为几条记录。

当一对一 Source -> Destination ETL，甚至是一对多 Source -> Destination 与 V3 中的新可枚举爆炸器时，Kiba 非常简单，但是我没有看到通往多对一 ETL 管道的清晰路径。

任何建议或指导将不胜感激。

【问题讨论】：

标签： ruby etl kiba-etl

【解决方案1】：

很高兴您发现 Kiba 很有用！此用例有多种解决方案。

我在这里做了一些假设（如果这些不正确，解决方案将存在，但会有所不同，例如边界检测和外部存储）：

您正在处理有限的批次（而不是连续的更新流）。
您提到的少数摘要记录可以保存在内存中。

我的建议是利用 Kiba v3 的能力在 transform 的 close 方法中产生记录（在 this article 中有更深入的描述）：

class InMemoryReduceTransform
  attr_reader :buffer, :summarize_cb

  def initialize(summarize_cb:)
    @buffer = []
    @summarize_cb = summarize_cb
  end

  def process(row)
    buffer << row
    nil # do not forward the row to the rest of the pipeline
  end

  def close
    summarize_cb(buffer).each do |row|
      yield row
    end
  end
end

本质上，您只需将输入行堆叠起来，直到源数据用完，此时将调用close 方法，然后汇总您拥有的数据并生成 N 个汇总行。

注意：这是一个简单的实现，可让您走上正轨。 Kiba Pro 的下一次迭代将包括一个更具可扩展性和通用性的版本（在供应商支持下）。有兴趣请联系！

如果这能正确回答您的问题，请告诉我！

【讨论】：

感谢@thibaut，这是完美的。我不知道我们也可以在变压器关闭期间屈服。回复：Kiba Pro - 我肯定会向我的客户提出这个问题。我们已经在使用 Sidekiq Pro，因此他们已经乐于投资优秀的商业开源项目了 :) 继续努力！
我刚刚重新阅读了这篇文章，这次更加专注，你完全回答了那里的问题。再次感谢:)
很高兴这解决了您的用例！ & 感谢您提高客户对 Kiba 的认识，非常感谢！