使用 kiba ETL 检查 CSV 文件标头的最佳位置

Question

我需要检查一下：

• 存在标题行
• 标头包含一组特定的标头

这样做的最佳地点是什么。我有一些可能的解决方案，但不知道更惯用的解决方案

• 在运行完整 ETL 之前检查，例如在 Kiba.parse 块之前
• 签入 ETL 内的 pre_process 块
• 检入 ETL 源。我更喜欢这个，因为它更易于重用（需要将必填字段作为参数传递）

请注意，即使我可以在 transform 块中检查 row 上可用的字段，此解决方案似乎也不是很有效，因为它会为每一行运行。

任何提示表示赞赏

Answer 1

实现这一点的方法多种多样，而且都是非常惯用的：

在源代码级别（传递标头数组）

您可以使用CSV 而不使用headers: true，这提供了仔细检查标题的机会：

class CSVSource
  def initialize(filename:, csv_options:, expected_headers:)
  # SNIP

  def each
    CSV.foreach(filename, csv_options).with_index do |row, file_row_index|
      if file_row_index == 0
        check_headers!(actual: row.to_a, expected: expected_headers)
        next # do not propagate the headers row
      else
        yield(Hash[expected_headers.zip(row.to_a)])
      end
    end
  end

  def check_headers!(actual:, expected:)
  # SNIP - verify uniqueness, presence, raise a clear message if needed
end     

在源代码级别（让调用者使用 lambda 定义行为）

class CSVSource
  def initialize(after_headers_read_callback:, ...)
    @after_headers_read_callback = ...

  def each
    CSV.foreach(filename, csv_options).with_index do |row, file_row_index|
      if file_row_index == 0
        @after_headers_read_callback.call(row.to_a)
        next
      end
      # ...
    end
  end

lambda 将让调用者定义他们自己的检查，在需要时引发等等，这样更便于重用。

在变换级别

如果您想进一步解耦组件（例如，将标题处理与行来自 CSV 源的事实分开），您可以使用转换。

我通常使用这种设计，这样可以更好地重用（这里使用 CSV 源，会产生一些元数据）：

def transform_array_rows_to_hash_rows(after_headers_read_callback:)
  transform do |row|
    if row.fetch(:file_row_index) == 0
      @headers = row.fetch(:row)
      after_headers_read_callback.call(@headers)
      nil
    else
      Hash[@headers.zip(row.fetch(:row))].merge(
        filename: row.fetch(:filename),
        file_row_index: row.fetch(:file_row_index)
      )
    end
  end
end

什么不推荐

在所有情况下，避免在Kiba.parse 本身中进行任何处理。这是一种更好的设计，可确保仅在您调用 Kiba.run 时才会发生 IO（因为它将更加面向未来，并且将在 Kiba 的更高版本中支持自省功能）。

另外，不推荐使用pre_process（虽然它会起作用），因为它会导致一些重复等。

希望这会有所帮助，如果不清楚，请告诉我！