编辑：现在是 2020 年，Kiba ETL v3 包含一个更好的方法来做到这一点。查看这篇文章https://thibautbarrere.com/2020/03/05/new-in-kiba-etl-v3 了解所有相关信息。
木场作者在这里！您可以通过多种不同的方式实现这一点，主要取决于数据大小和您的实际需求。以下是几种可能性。
在 Kiba 脚本中使用变量进行聚合
require 'awesome_print'
transform do |r|
r[:amount] = BigDecimal.new(r[:amount])
r
end
total_amounts = Hash.new(0)
transform do |r|
total_amounts[r[:name]] += r[:amount]
r
end
post_process do
# pretty print here, but you could save to a CSV too
ap total_amounts
end
这是最简单的方法，但也很灵活。
不过，它会将您的聚合保存在内存中，因此这可能足够好，也可能不够好，具体取决于您的场景。请注意，目前 Kiba 是单线程的（但“Kiba Pro”将是多线程的），因此目前不需要为聚合添加锁或使用线程安全结构。
从 post_process 块调用 TextQL
另一种快速简便的聚合方法是先生成一个非聚合的 CSV 文件，然后利用 TextQl 实际进行聚合，如下所示：
destination CsvSource, 'non-aggregated-output.csv', [:name, :amount]
post_process do
query = <<SQL
select
name,
/* apparently sqlite has reduced precision, round to 2 for now */
round(sum(amount), 2) as total_amount
from tbl group by name
SQL
textql('non-aggregated-output.csv', query, 'aggregated-output.csv')
end
定义了以下助手：
def system!(cmd)
raise "Failed to run command #{command}" unless system(command)
end
def textql(source_file, query, output_file)
system! "cat #{source_file} | textql -header -output-header=true -sql \"#{query}\" > #{output_file}"
# this one uses csvfix to pretty print the table
system! "cat #{output_file} | csvfix ascii_table"
end
在进行计算时要注意精度。
编写内存聚合目的地
在这里可以使用的一个有用技巧是用一个类包装给定的目的地来进行聚合。下面是它的样子：
class InMemoryAggregate
def initialize(sum:, group_by:, destination:)
@aggregate = Hash.new(0)
@sum = sum
@group_by = group_by
# this relies a bit on the internals of Kiba, but not too much
@destination = destination.shift.new(*destination)
end
def write(row)
# do not write, but count here instead
@aggregate[row[@group_by]] += row[@sum]
end
def close
# use close to actually do the writing
@aggregate.each do |k,v|
# reformat BigDecimal additions here
value = '%0.2f' % v
@destination.write(@group_by => k, @sum => value)
end
@destination.close
end
end
你可以这样使用：
# convert your string into an actual number
transform do |r|
r[:amount] = BigDecimal.new(r[:amount])
r
end
destination CsvDestination, 'non-aggregated.csv', [:name, :amount]
destination InMemoryAggregate,
sum: :amount, group_by: :name,
destination: [
CsvDestination, 'aggregated.csv', [:name, :amount]
]
post_process do
system!("cat aggregated.csv | csvfix ascii_table")
end
这个版本的好处是您可以将聚合器重复用于不同的目的地（如数据库或其他任何地方）。
请注意，这会将所有聚合保留在内存中，就像第一个版本一样。
插入具有聚合功能的商店
另一种方法（如果您的数据量很大，则特别有用）是将生成的数据发送到能够为您聚合数据的东西中。它可以是常规的 SQL 数据库、Redis 或任何更花哨的东西，然后您可以根据需要进行查询。
正如我所说，实施将在很大程度上取决于您的实际需求。希望您能在这里找到适合您的东西！