有没有办法在 Kiba 作业结束时返回一些数据？

Question

如果有一种方法可以从 Kiba ETL 运行中获取某种返回对象，这样我就可以使用其中的数据返回有关管道运行情况的报告。

我们有一个每 10 分钟运行一次的作业，平均处理 20 到 50k 条记录，并将它们压缩为摘要记录，其中一些是创建的，而一些是更新的。问题是，如果不翻阅大量日志文件，就很难知道发生了什么，显然，日志对最终用户也很有用。

有没有办法在管道运行时用任意数据填充某种结果对象？例如

• 在源代码中找到 25.7k 行
• 此转换器丢弃了 782 条记录
• 已插入 100 条记录
• 更新了 150 条记录
• 20 条记录有错误（在这里）
• 此记录的统计量最高
• 1200 条记录属于此 VIP 客户
• 等

最后，使用该数据发送电子邮件摘要、填充网页、呈现一些控制台输出等。

目前，我现在可以看到这个工作的唯一方法是在设置期间发送一个对象，并在它流经源、转换器和目标时对其进行变异。运行完成后，随后检查变量并对现在存在的数据进行处理。

这是应该怎么做，还是有更好的方法？

编辑

只想补充一点，我不想在post_process 块中处理这个问题，因为管道通过许多不同的媒介使用，我希望每个用例都处理自己的反馈机制。 ETL 管道也更清洁 (imo)，不必担心它在哪里使用，以及该使用场景的反馈期望是什么......

Answer 1

答案在很大程度上取决于上下文，但这里有一些指导原则。

如果结果对象不是太大，我确实建议您传递一个空的结果对象（通常是 Hash），然后在运行期间填充它（您也可以使用某种形式的中间件来跟踪异常本身）。

您将如何填充它取决于上下文和您的实际需求，但这可以以与工作无关的方式完成（也许使用 DSL 扩展 https://github.com/thbar/kiba/wiki/How-to-extend-the-Kiba-DSL，您可以实现一些相当高级的扩展，这些扩展将注册所需的转换或块来实现您所需要的）。

该对象可以按原样使用，也可以序列化为 JSON 或类似的，如果您以后需要提供一些丰富的输出（或者您可以使用它来准备其他东西），甚至可以将其存储到数据库中。

如果需要，您甚至可以为此目的在特定数据库中建立一些结构合理的内容（例如，如果您需要一种简单的方法将其公开给客户）。

请注意，您可以以编程方式定义post_process，而工作却没有意识到它（没有耦合）。这是一个非常简单的例子：

module ETL
  module DSLExtensions
    module EmailReport
      def setup_email_report
        pre_process do
          @email_report_stats = Hash.new(0)
        end

        post_process do
          # Do the actual email sending
        end
      end

      def track_event!(event:)
        @email_report_stats[event] += 1
      end
    end
  end
end

Kiba.parse do
  extend ETL::DSLExtensions::EmailReport

  # this will register the pre/post process
  setup_email_report

  source ...

  track_event!(event: 'row_read')

  transform
  transform
  transform

  track_event!(event: 'row_written')

  destination ...
end

如果您这样做，请确保使用非常好的命名空间变量，以避免任何冲突。

请注意，如前所述，这不包括失败的情况，但你明白了！