ETL (Extract-Transform-Load) with Kiba（2）

franket

发布于 2021-10-18 11:57:11

4550

创建一个库文件

我们采用尽量模块化的思想，将可重用的代码集中放到一个库文件中(common.rb)以便于维护，核心逻辑放到主文件中(convert-csv.etl)

加入对 CSV 源的定义

[root@h102 kiba]# vim common.rb
[root@h102 kiba]# cat common.rb 
require 'csv'

class CsvSource
  def initialize(file, options)
    @file = file
    @options = options
  end
  
  def each
    CSV.foreach(@file, @options) do |row|
      yield row.to_hash
    end
  end
end
[root@h102 kiba]#

Note: 随着项目的累积与扩展，会产生各种各样的源定义，为了便于维护也可以将这些源定义分离出来成为单独的文件

对 CSV 源进行测试

[root@h102 kiba]# vim commandes.csv
[root@h102 kiba]# cat commandes.csv 
date_facture;montant_eur;numero_commande
7/3/2015;10,96;FA1986
7/3/2015;85,11;FA1987
8/3/2015;6,41;FA1988
[root@h102 kiba]# vim convert-csv.etl 
[root@h102 kiba]# cat convert-csv.etl 
require_relative 'common'

# read from source CSV file
source CsvSource, 'commandes.csv', col_sep: ';', headers: true, header_converters: :symbol
[root@h102 kiba]# bundle exec kiba convert-csv.etl 
[root@h102 kiba]# 
[root@h102 kiba]#

require_relative 和 require 所起的功能一样，只是引用文件的位置为自身的相对位置而与 $LAOD_PATH ($:) 路径无关

从对 CSV 源的定义我们知道，'commandes.csv' 被初始化给了 @file ，而 col_sep: ';', headers: true, header_converters: :symbol 被初始化给了 @options

参数的意思就是：使用 CSV 打开 ‘commandes.csv’ 文件，这个文件是以 ‘;’ 作为字段分割符的，有头信息，将头信息转化为 ‘:symbol’ 的形式

Tip: CSV 是标准库，其使用方法与相关细节可以参考 CSV gem

最后的执行结果并没有报加载异常，表明代码可以正常执行

打印结果

我们可以将中间结果打印出来，以方便调试

在 common.rb 中定义一个 show_me 方法，加入一段显示逻辑，然后运行kiba项目

[root@h102 kiba]# vim common.rb 
[root@h102 kiba]# cat common.rb 
require 'csv'

class CsvSource
  def initialize(file, options)
    @file = file
    @options = options
  end
  
  def each
    CSV.foreach(@file, @options) do |row|
      yield row.to_hash
    end
  end
end


require 'awesome_print'

def show_me
  transform do |row|
    ap row
    row # always return the row to keep it in the pipeline
  end
end

[root@h102 kiba]# ls
commandes.csv  common.rb  convert-csv.etl  Gemfile  Gemfile.lock
[root@h102 kiba]# vim convert-csv.etl 
[root@h102 kiba]# cat convert-csv.etl 
require_relative 'common'

# read from source CSV file
source CsvSource, 'commandes.csv', col_sep: ';', headers: true, header_converters: :symbol

show_me
[root@h102 kiba]# bundle exec kiba convert-csv.etl
{
       :date_facture => "7/3/2015",
        :montant_eur => "10,96",
    :numero_commande => "FA1986"
}
{
       :date_facture => "7/3/2015",
        :montant_eur => "85,11",
    :numero_commande => "FA1987"
}
{
       :date_facture => "8/3/2015",
        :montant_eur => "6,41",
    :numero_commande => "FA1988"
}
[root@h102 kiba]#

现在已经可以成功解析使用 ';' 分割的字段，并且以hash的形式打印出来

解析数值

加入解析数值的类 ParseFrenchFloat ，并定义处理逻辑

[root@h102 kiba]# vim common.rb 
[root@h102 kiba]# cat common.rb 
require 'csv'

class CsvSource
  def initialize(file, options)
    @file = file
    @options = options
  end
  
  def each
    CSV.foreach(@file, @options) do |row|
      yield row.to_hash
    end
  end
end


require 'awesome_print'

def show_me
  transform do |row|
    ap row
    row # always return the row to keep it in the pipeline
  end
end


class ParseFrenchFloat
  def initialize(from:, to:)
    @from = from
    @to = to
  end
  
  def process(row)
    row[@to] = Float(row[@from].gsub(',', '.'))
    row
  end
end

[root@h102 kiba]# vim convert-csv.etl 
[root@h102 kiba]# cat convert-csv.etl 
require_relative 'common'

# read from source CSV file
source CsvSource, 'commandes.csv', col_sep: ';', headers: true, header_converters: :symbol

# Parse the numbers
transform ParseFrenchFloat, from: :montant_eur, to: :amount_eur

# show details of row contents
show_me
[root@h102 kiba]# bundle exec kiba convert-csv.etl
{
       :date_facture => "7/3/2015",
        :montant_eur => "10,96",
    :numero_commande => "FA1986",
         :amount_eur => 10.96
}
{
       :date_facture => "7/3/2015",
        :montant_eur => "85,11",
    :numero_commande => "FA1987",
         :amount_eur => 85.11
}
{
       :date_facture => "8/3/2015",
        :montant_eur => "6,41",
    :numero_commande => "FA1988",
         :amount_eur => 6.41
}
[root@h102 kiba]#

其中最主要的就是 row[@to] = Float(row[@from].gsub(',', '.'))

它的意思就是对 from 字段(或 Key) 指向的值进行处理，将其中的 , 替换为 .，然后使用 Float 转化为浮点数，然后赋予给 to 字段，这个字段是新字段，在 row hash 中添加入新的 KV 对

运行的结果正如预期

本文系转载，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

文件存储