Added etl_tstamp into Scala Hadoop Enrich (#396)

snowplow · Jun 25, 2014 · 9354cc4 · 9354cc4
1 parent d55b07d
commit 9354cc4
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/3-enrich/emr-etl-runner/lib/snowplow-emr-etl-runner/emr_job.rb b/3-enrich/emr-etl-runner/lib/snowplow-emr-etl-runner/emr_job.rb
@@ -47,7 +47,9 @@ def initialize(debug, shred, s3distcp, config)
 
         # Configuration
         assets = self.class.get_assets(config[:s3][:buckets][:assets], config[:etl][:versions][:hadoop_enrich], config[:etl][:versions][:hadoop_shred])
-        run_id = Time.new.strftime("%Y-%m-%d-%H-%M-%S")
+        run_tstamp = Time.new
+        run_id = run_tstamp.strftime("%Y-%m-%d-%H-%M-%S")
+        etl_tstamp = run_tsta.to_i
 
         # Create a job flow with your AWS credentials
         @jobflow = Elasticity::JobFlow.new(config[:aws][:access_key_id], config[:aws][:secret_access_key])
@@ -159,6 +161,7 @@ def initialize(debug, shred, s3distcp, config)
             :errors => self.class.fix_equals(self.class.partition_by_run(csbe[:errors], run_id, config[:etl][:continue_on_unexpected_error]))
           },
           { :input_format     => config[:etl][:collector_format],
+            :etl_tstamp       => etl_tstamp,
             :maxmind_file     => assets[:maxmind],
             :anon_ip_octets   => self.class.get_anon_ip_octets(config[:enrichments][:anon_ip])
           }

diff --git a/...oop-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich.hadoop/EtlJobConfig.scala b/...oop-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich.hadoop/EtlJobConfig.scala
@@ -110,8 +110,9 @@ object EtlJobConfig {
     val outFolder = args.requiredz("output_folder")
     val badFolder = args.requiredz("bad_rows_folder")
     val anonOctets = args.requiredz("anon_ip_octets").flatMap(q => getAnonOctets(q))
+    val etlTstamp = args.requiredz("etl_tstamp").flatMap(t => extractTimestamp("etl_tstamp", t))
     val exceptionsFolder = args.optionalz("exceptions_folder")
 
-    (inFolder.toValidationNel |@| inFormat.toValidationNel |@| maxmindFile.toValidationNel |@| outFolder.toValidationNel |@| badFolder.toValidationNel |@| anonOctets.toValidationNel |@| exceptionsFolder.toValidationNel) { EtlJobConfig(_,_,_,_,_,_,_) }
+    (inFolder.toValidationNel |@| inFormat.toValidationNel |@| maxmindFile.toValidationNel |@| outFolder.toValidationNel |@| badFolder.toValidationNel |@| anonOctets.toValidationNel |@| etlTstamp.toValidationNel |@| exceptionsFolder.toValidationNel) { EtlJobConfig(_,_,_,_,_,_,_,_) }
   }
-}
+}