initial.rb

source :idx, 
{
    :file => 'listings.txt',
    :parser => :delimited,
    :skip_lines => 1
},
[
    :listing_id,
    :area,
    :baths_full,
    :baths_part,
    :baths,
    :bdrms,
    :dt_add,
    :l_ag1_fax,
    :l_ag1_id,
    :l_ag1_name,
    :l_ag1_phone,
    :l_off_fax,
    :l_off_id,
    :l_off_name,
    :l_off_phone,
    :mls_number,
    :photo_no,
    :photo_yn,
    :price_current,
    :prop_type,
    :status,
    :status_flag,
    :virtual_tour,
    :areaname,
    :internet_remarks,
    :style,
    :sub_style,
    :subarea,
    :locale,
    :bsmt_desc,
    :zip
]

transform :listing_id, :type, :type => :integer
transform :mls_number, :type, :type => :integer
transform :bdrms,      :type, :type => :integer
transform :l_ag1_id,   :type, :type => :integer

transform(:l_ag1_name) do |name, value, row|
    value.scan(/[^,]/).join
end

transform(:l_off_name) do |name, value, row|
    value.scan(/[^,]/).join
end

transform(:areaname) do |name, value, row|
    value.scan(/[^,]/).join
end

transform(:internet_remarks) do |name, value, row|
    '"%s"' % value
end


destination :town_dimension,
{
    :file => 'intermediate/town_dimension.1.txt'
},
{
    :order => [:areaname,
               :zip]
}

destination :property_dimension,
{
    :file => 'intermediate/property_dimension.1.txt'
},
{
    :order => [:listing_id, 
               :mls_number, 
               :bdrms,
               :price_current,
               :dt_add,
               :zip]
}

destination :office_dimension,
{
    :file => 'intermediate/office_dimension.1.txt'
},
{
    :order => [:l_off_id,
               :l_off_name,
               :l_off_phone,
               :l_off_fax]
}

destination :agent_dimension,
{
    :file => 'intermediate/agent_dimension.1.txt'
},
{
    :order => [:l_ag1_id,
               :l_ag1_name,
               :l_ag1_phone,
               :l_off_id]
}

destination :listing_event_facts,
{
    :file => 'intermediate/listing_event_facts.1.txt'
},
{
    :order => [:listing_id,
               :l_ag1_id,
               :price_current,
               :internet_remarks]
}

listing_event_fact.rb

infile  = 'intermediate/listing_event_facts.1.txt'
outfile = 'intermediate/listing_event_facts.2.txt'
target_table = 'listing_event_facts'

source :listing_event_facts, 
{
    :file => infile,
    :parser => :delimited
},
[
    :listing_id,
    :agent_id,
    :price,
    :description
]

transform :price, :type, :type => :integer

rename :listing_id, :property_dimension_id
transform :property_dimension_id, 
    :foreign_key_lookup, 
    :resolver => SQLResolver.new('property_dimension', 
                                 'listing_id', 
                                 :data_warehouse),
    :default  => nil

rename :agent_id, :agent_dimension_id
transform :agent_dimension_id, 
    :foreign_key_lookup, 
    :resolver => SQLResolver.new('agent_dimension', 
                                 'agent_id', 
                                 :data_warehouse),
    :default  => nil

before_write :check_exist,
    :target  => :data_warehouse,
    :table   => target_table,
    :columns => [:price, 
                 :property_dimension_id,
                 :agent_dimension_id]

destination :listing_event_facts,
{
    :file => outfile
},
{
    :order => [:price, 
               :description,
               :property_dimension_id,
               :agent_dimension_id],
}

post_process :bulk_import, {
  :file => outfile,
  :target => :data_warehouse,
  :table => target_table,
  :columns => [:price, 
               :description,
               :property_dimension_id,
               :agent_dimension_id]
}

property_dimension.rb

infile  = 'intermediate/property_dimension.1.txt'
outfile = 'intermediate/property_dimension.2.txt'
target_table = 'property_dimension'

source :property_dimension, 
{
    :file => infile,
    :parser => :delimited
},
[
    :listing_id, 
    :mls_number, 
    :bedrooms,
    :price,
    :date_listed,
    :zip
]

transform :listing_id, :type, :type => :integer
transform :mls_number, :type, :type => :integer
transform :bedrooms,   :type, :type => :integer

rename :zip, :town_dimension_id
transform :town_dimension_id, 
    :foreign_key_lookup, 
    :resolver => SQLResolver.new('town_dimension', 'zip', :data_warehouse),
    :default  => nil

before_write :check_unique, :keys => [:listing_id]

before_write :surrogate_key, 
    :target => :data_warehouse, 
    :table  => target_table, 
    :column => 'id'

before_write :check_exist,
    :target => :data_warehouse,
    :table  => target_table,
    :columns => [:listing_id]

destination :property_dimension,
{
    :file => outfile
},
{
    :order => [:id,
               :listing_id, 
               :mls_number, 
               :bedrooms,
               :town_dimension_id],
    :virtual => { :id => :surrogate_key }
}

post_process :bulk_import, {
  :file => outfile,
  :target => :data_warehouse,
  :table => target_table,
  :columns => [:id, 
             :listing_id, 
             :mls_number, 
             :bedrooms,
             :town_dimension_id]
}