Merge pull request #8 from inside-track/Aggregate

Added support for generic aggregation in dataframes.

Author: Sterling Paramore

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    remi (0.2.1)
+    remi (0.2.3)
       activesupport (~> 4.2)
       bond (~> 0.5)
       cucumber (~> 2.1)

lib/remi.rb

@@ -18,6 +18,8 @@
 require 'active_support/core_ext/object/try'
 require 'active_support/core_ext/object/inclusion'
 require 'active_support/core_ext/string/inflections'
+require 'active_support/core_ext/string/strip'
+require 'active_support/core_ext/string/filters'
 require 'active_support/core_ext/numeric/time'
 require 'active_support/core_ext/numeric/conversions'
 require 'active_support/core_ext/date/calculations'
@@ -30,7 +32,6 @@
 
 # Remi
 require 'remi/version.rb'
-require 'remi/core/string.rb'
 
 require 'remi/settings'
 require 'remi/job'
@@ -39,8 +40,8 @@
 require 'remi/data_subject'
 require 'remi/sf_bulk_helper' # separate into SF support package
 
-require 'remi/core/daru'
-require 'remi/core/refinements'
+require 'remi/refinements/symbolizer'
+require 'remi/refinements/daru'
 
 require 'remi/extractor/sftp_file'
 

lib/remi/cli.rb

@@ -15,7 +15,7 @@ def parse(args = ARGV)
       options = {}
 
       opt_parser = OptionParser.new do |opts|
-        opts.banner = <<-EOT.unindent
+        opts.banner = <<-EOT.strip_heredoc
           Usage: Command line helpers for Remi.
         EOT
 

lib/remi/core/daru.rb

@@ -1,5 +1,5 @@
 module Remi::BusinessRules
-  using Remi::Core::Refinements
+  using Remi::Refinements::Symbolizer
 
   def self.parse_full_field(full_field_name)
     full_field_name.split(':').map(&:strip)
@@ -230,6 +230,15 @@ def size
       @data_obj.df.size
     end
 
+    # Public: Converts the data subject to a hash where the keys are the table
+    # columns and the values are an array for the value of column for each row.
+    def column_hash
+      @data_obj.df.to_hash.reduce({}) do |h, (k,v)|
+        h[k.symbolize] = v.to_a
+        h
+      end
+    end
+
     # For debugging only
     def _df
       @data_obj.df
@@ -429,6 +438,17 @@ def to_df(seed_hash, field_symbolizer:)
       end
       df
     end
+
+    # Public: Converts a Cucumber::Ast::Table to a hash where the keys are the table
+    # columns and the values are an array for the value of column for each row.
+    def column_hash
+      @table.hashes.reduce({}) do |h, row|
+        row.each do |k,v|
+          (h[k.symbolize] ||= []) << v
+        end
+        h
+      end
+    end
   end
 
 

lib/remi/core/string.rb

@@ -0,0 +1,23 @@
+Feature: Tests the aggregate refinement to the Daru library
+
+  Background:
+    Given the job is 'Aggregate'
+    And the job source 'Source Data'
+    And the job target 'Target Data'
+
+    And the source 'Source Data'
+    And the target 'Target Data'
+
+  Scenario: The aggregator should find the minimum year for each 'Alpha'
+    Given the following example record for 'Source Data':
+      | Alpha | Year | something |
+      | a     | 2016 | 1 |
+      | a     | 2018 | 1 |
+      | b     | 2016 | 2 |
+      | b     | 2010 | 3 |
+      | a     | 2017 | 4 |
+    And the following example record called 'expected result':
+      | Alpha | Year |
+      | a     | 2016 |
+      | b     | 2010 |
+    Then the target should match the example 'expected result'

lib/remi/cucumber/business_rules.rb

@@ -201,6 +201,22 @@
   step "the target field '#{target_field}' is set to the value \"*#{date_reference}*\""
 end
 
+Then /^the target '(.+)' should match the example '([[:alnum:]\s]+)'$/ do |target_name, example_name|
+  @brt.run_transforms
+
+  target_hash = @brt.targets[target_name].column_hash
+  example_hash = @brt.examples[example_name].column_hash
+  common_keys = target_hash.keys & example_hash.keys
+
+  expect(target_hash.select { |k,v| common_keys.include? k })
+    .to eq example_hash.select { |k,v| common_keys.include? k }
+end
+
+Then /^the target should match the example '([[:alnum:]\s]+)'$/ do |example_name|
+  target_name = @brt.targets.keys.first
+  step "the target '#{target_name}' should match the example '#{example_name}'"
+end
+
 
 ### Transforms
 

lib/remi/project/features/aggregate.feature

@@ -1,3 +1,4 @@
 require_relative '../../jobs/sample_job'
 require_relative '../../jobs/copy_source_job'
 require_relative '../../jobs/transforms/transform_jobs'
+require_relative '../../jobs/aggregate_job'

lib/remi/project/features/step_definitions/remi_step.rb

@@ -0,0 +1,20 @@
+require_relative 'all_jobs_shared'
+
+class AggregateJob
+  include AllJobsShared
+  using Remi::Refinements::Daru
+
+  define_source :source_data, Remi::DataSource::DataFrame
+  define_target :target_data, Remi::DataTarget::DataFrame
+
+  define_transform :main, sources: :source_data, targets: :target_data do
+
+    mymin = lambda do |field, df, indicies|
+      values = indicies.map { |idx| df.row[idx][field] }
+      values.min
+    end
+
+    target_data.df = source_data.df.aggregate(by: :alpha, func: mymin.curry.(:year)).detach_index
+    target_data.df.vectors = Daru::Index.new([:alpha, :year])
+  end
+end