lib/etl/engine.rb
def process_control(control)
      control = ETL::Control::Control.resolve(control)
      say_on_own_line "Processing control #{control.file}"
      
      ETL::Engine.job = ETL::Execution::Job.new.tap do |job|
        job.control_file = control.file
        job.status = 'executing'
        job.batch_id = ETL::Engine.batch ? ETL::Engine.batch.id : nil
        job.save!
      end

      execute_dependencies(control)
      
      start_time = Time.now
      pre_process(control)
      sources = control.sources
      destinations = control.destinations
      
      say "Skipping bulk import" if Engine.skip_bulk_import
      
      sources.each do |source|
        Engine.current_source = source
        Engine.logger.debug "Processing source #{source.inspect}"
        say "Source: #{source}"
        say "Limiting enabled: #{Engine.limit}" if Engine.limit != nil
        say "Offset enabled: #{Engine.offset}" if Engine.offset != nil
        source.each_with_index do |row, index|
          # Break out of the row loop if the +Engine.limit+ is specified and 
          # the number of rows read exceeds that value.
          if Engine.limit != nil && Engine.rows_read >= Engine.limit
            puts "Reached limit of #{Engine.limit}"
            break
          end
          
          Engine.logger.debug "Row #{index}: #{row.inspect}"
          Engine.rows_read += 1
          Engine.current_source_row = index + 1
          say_without_newline "." if Engine.realtime_activity && index > 0 && index % 1000 == 0
          
          # At this point a single row may be turned into multiple rows via row 
          # processors all code after this line should work with the array of 
          # rows rather than the single row
          rows = [row]
          
          t = Benchmark.realtime do
            begin
              Engine.logger.debug "Processing after read"
              control.after_read_processors.each do |processor|
                processed_rows = []
                rows.each do |row|
                  processed_rows << processor.process(row) unless empty_row?(row)
                end
                rows = processed_rows.flatten.compact
              end
            rescue => e
              msg = "Error processing rows after read from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              # TODO - track more information: row if possible, full exception...
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:after_reads] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              Engine.logger.debug "Executing transforms"
              rows.each do |row|
                # only do the transform if there is a row
                unless empty_row?(row)
                  control.transforms.each do |transform|
                    name = transform.name.to_sym
                    row[name] = transform.transform(name, row[name], row)
                  end
                end
              end
            rescue ResolverError => e
              Engine.logger.error(e.message)
              track_error(control, e.message)
            rescue => e
              msg = "Error transforming from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
            ensure
              begin
                exceeded_error_threshold?(control) ? break : next
              rescue => inner_error
                puts inner_error
              end
            end
          end
          benchmarks[:transforms] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              # execute row-level "before write" processing
              Engine.logger.debug "Processing before write"
              control.before_write_processors.each do |processor|
                processed_rows = []
                rows.each do |row|
                  processed_rows << processor.process(row) unless empty_row?(row)
                end
                rows = processed_rows.flatten.compact
              end
            rescue => e
              msg = "Error processing rows before write from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:before_writes] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              # write the row to the destination
              destinations.each_with_index do |destination, index|
                Engine.current_destination = destination
                rows.each do |row|
                  destination.write(row)
                  Engine.rows_written += 1 if index == 0
                end
              end
            rescue => e
              msg = "Error writing to #{Engine.current_destination}: #{e}"
              track_error(control, msg)
              Engine.logger.error msg
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:writes] += t unless t.nil?
        end
        
        if exceeded_error_threshold?(control)
          say_on_own_line "Exiting due to exceeding error threshold: #{control.error_threshold}"
          ETL::Engine.exit_code = 1
        end
        
      end
      
      destinations.each do |destination|
        destination.close
      end
      
      say_on_own_line "Executing before post-process screens"
      begin
        execute_screens(control)
      rescue FatalScreenError => e
        say "Fatal screen error during job execution: #{e.message}"
        ETL::Engine.exit_code = 2
      rescue ScreenError => e
        say "Screen error during job execution: #{e.message}"
        return
      else
        say "Screens passed"
      end
      
      post_process(control)
      
      if sources.length > 0
        say_on_own_line "Read #{Engine.rows_read} lines from sources"
      end
      if destinations.length > 0
        say "Wrote #{Engine.rows_written} lines to destinations"
      end

      say_on_own_line "Executing after post-process screens"
      begin
        execute_screens(control, :after_post_process)
      rescue FatalScreenError => e
        say "Fatal screen error during job execution: #{e.message}"
        ETL::Engine.exit_code = 3
      rescue ScreenError => e
        say "Screen error during job execution: #{e.message}"
        return
      else
        say "Screens passed"
      end

      say_on_own_line "Completed #{control.file} in #{distance_of_time_in_words(start_time)} with #{errors.length} errors."
      say "Processing average: #{Engine.average_rows_per_second} rows/sec)"
      
      say "Avg after_reads: #{Engine.rows_read/benchmarks[:after_reads]} rows/sec" if benchmarks[:after_reads] > 0
      say "Avg before_writes: #{Engine.rows_read/benchmarks[:before_writes]} rows/sec" if benchmarks[:before_writes] > 0
      say "Avg transforms: #{Engine.rows_read/benchmarks[:transforms]} rows/sec" if benchmarks[:transforms] > 0
      say "Avg writes: #{Engine.rows_read/benchmarks[:writes]} rows/sec" if benchmarks[:writes] > 0

      # say "Avg time writing execution records: #{ETL::Execution::Record.average_time_spent}"
      # 
      # ETL::Transform::Transform.benchmarks.each do |klass, t|
#         say "Avg #{klass}: #{Engine.rows_read/t} rows/sec"
#       end

      ActiveRecord::Base.verify_active_connections!
      ETL::Engine.job.completed_at = Time.now
      ETL::Engine.job.status = (errors.length > 0 ? 'completed with errors' : 'completed')
      ETL::Engine.job.save!
    end
def process_control(control)
      control = ETL::Control::Control.resolve(control)
      say_on_own_line "Processing control #{control.file}"
      
      ETL::Engine.job = ETL::Execution::Job.new.tap do |job|
        job.control_file = control.file
        job.status = 'executing'
        job.batch_id = ETL::Engine.batch ? ETL::Engine.batch.id : nil
        job.save!
      end

      execute_dependencies(control)
      
      start_time = Time.now
      pre_process(control)
      sources = control.sources
      destinations = control.destinations
      
      say "Skipping bulk import" if Engine.skip_bulk_import
      
      sources.each do |source|
        Engine.current_source = source
        Engine.logger.debug "Processing source #{source.inspect}"
        say "Source: #{source}"
        say "Limiting enabled: #{Engine.limit}" if Engine.limit != nil
        say "Offset enabled: #{Engine.offset}" if Engine.offset != nil
        source.each_with_index do |row, index|
          # Break out of the row loop if the +Engine.limit+ is specified and 
          # the number of rows read exceeds that value.
          if Engine.limit != nil && Engine.rows_read >= Engine.limit
            puts "Reached limit of #{Engine.limit}"
            break
          end
          
          Engine.logger.debug "Row #{index}: #{row.inspect}"
          Engine.rows_read += 1
          Engine.current_source_row = index + 1
          say_without_newline "." if Engine.realtime_activity && index > 0 && index % 1000 == 0
          
          # At this point a single row may be turned into multiple rows via row 
          # processors all code after this line should work with the array of 
          # rows rather than the single row
          rows = [row]
          
          t = Benchmark.realtime do
            begin
              Engine.logger.debug "Processing after read"
              control.after_read_processors.each do |processor|
                processed_rows = []
                rows.each do |row|
                  processed_rows << processor.process(row) unless empty_row?(row)
                end
                rows = processed_rows.flatten.compact
              end
            rescue => e
              msg = "Error processing rows after read from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              # TODO - track more information: row if possible, full exception...
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:after_reads] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              Engine.logger.debug "Executing transforms"
              rows.each do |row|
                # only do the transform if there is a row
                unless empty_row?(row)
                  control.transforms.each do |transform|
                    name = transform.name.to_sym
                    row[name] = transform.transform(name, row[name], row)
                  end
                end
              end
            rescue ResolverError => e
              Engine.logger.error(e.message)
              track_error(control, e.message)
            rescue => e
              msg = "Error transforming from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
            ensure
              begin
                exceeded_error_threshold?(control) ? break : next
              rescue => inner_error
                puts inner_error
              end
            end
          end
          benchmarks[:transforms] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              # execute row-level "before write" processing
              Engine.logger.debug "Processing before write"
              control.before_write_processors.each do |processor|
                processed_rows = []
                rows.each do |row|
                  processed_rows << processor.process(row) unless empty_row?(row)
                end
                rows = processed_rows.flatten.compact
              end
            rescue => e
              msg = "Error processing rows before write from #{Engine.current_source} on line #{Engine.current_source_row}: #{e}"
              track_error(control, msg)
              Engine.logger.error(msg)
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:before_writes] += t unless t.nil?
          
          t = Benchmark.realtime do
            begin
              # write the row to the destination
              destinations.each_with_index do |destination, index|
                Engine.current_destination = destination
                rows.each do |row|
                  destination.write(row)
                  Engine.rows_written += 1 if index == 0
                end
              end
            rescue => e
              msg = "Error writing to #{Engine.current_destination}: #{e}"
              track_error(control, msg)
              Engine.logger.error msg
              e.backtrace.each { |line| Engine.logger.error(line) }
              exceeded_error_threshold?(control) ? break : next
            end
          end
          benchmarks[:writes] += t unless t.nil?
        end
        
        if exceeded_error_threshold?(control)
          say_on_own_line "Exiting due to exceeding error threshold: #{control.error_threshold}"
          ETL::Engine.exit_code = 1
        end
        
      end
      
      destinations.each do |destination|
        destination.close
      end
      
      say_on_own_line "Executing before post-process screens"
      begin
        execute_screens(control)
      rescue FatalScreenError => e
        say "Fatal screen error during job execution: #{e.message}"
        ETL::Engine.exit_code = 2
      rescue ScreenError => e
        say "Screen error during job execution: #{e.message}"
        return
      else
        say "Screens passed"
      end
      
      post_process(control)
      
      if sources.length > 0
        say_on_own_line "Read #{Engine.rows_read} lines from sources"
      end
      if destinations.length > 0
        say "Wrote #{Engine.rows_written} lines to destinations"
      end

      say_on_own_line "Executing after post-process screens"
      begin
        execute_screens(control, :after_post_process)
      rescue FatalScreenError => e
        say "Fatal screen error during job execution: #{e.message}"
        ETL::Engine.exit_code = 3
      rescue ScreenError => e
        say "Screen error during job execution: #{e.message}"
        return
      else
        say "Screens passed"
      end

      say_on_own_line "Completed #{control.file} in #{distance_of_time_in_words(start_time)} with #{errors.length} errors."
      say "Processing average: #{Engine.average_rows_per_second} rows/sec)"
      
      say "Avg after_reads: #{Engine.rows_read/benchmarks[:after_reads]} rows/sec" if benchmarks[:after_reads] > 0
      say "Avg before_writes: #{Engine.rows_read/benchmarks[:before_writes]} rows/sec" if benchmarks[:before_writes] > 0
      say "Avg transforms: #{Engine.rows_read/benchmarks[:transforms]} rows/sec" if benchmarks[:transforms] > 0
      say "Avg writes: #{Engine.rows_read/benchmarks[:writes]} rows/sec" if benchmarks[:writes] > 0

      # say "Avg time writing execution records: #{ETL::Execution::Record.average_time_spent}"
      # 
      # ETL::Transform::Transform.benchmarks.each do |klass, t|
#         say "Avg #{klass}: #{Engine.rows_read/t} rows/sec"
#       end

      ActiveRecord::Base.verify_active_connections!
      ETL::Engine.job.completed_at = Time.now
      ETL::Engine.job.status = (errors.length > 0 ? 'completed with errors' : 'completed')
      ETL::Engine.job.save!
    end
Process the specified control file