Scala Common Enrich: stored etl_tstamp in new field in CanonicalOutput (

#818)

Updated tests with etl_tstamp

3-enrich/scala-common-enrich/project/BuildSettings.scala

@@ -21,7 +21,7 @@ object BuildSettings {
   // Basic settings for our app
   lazy val basicSettings = Seq[Setting[_]](
     organization          :=  "com.snowplowanalytics",
-    version               :=  "0.4.0",
+    version               :=  "0.5.0-SNAPSHOT",
     description           :=  "Common functionality for enriching raw Snowplow events",
     scalaVersion          :=  "2.10.1",
     scalacOptions         :=  Seq("-deprecation", "-encoding", "utf8",

3-enrich/scala-common-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/EnrichmentManager.scala

@@ -61,7 +61,7 @@ object EnrichmentManager {
    *         either failure Strings or a
    *         NonHiveOutput.
    */
-  def enrichEvent(geo: IpGeo, hostEtlVersion: String, anonOctets: AnonOctets, raw: CanonicalInput): ValidatedCanonicalOutput = {
+  def enrichEvent(geo: IpGeo, hostEtlVersion: String, anonOctets: AnonOctets, etlTstamp: String, raw: CanonicalInput): ValidatedCanonicalOutput = {
 
     // Placeholders for where the Success value doesn't matter.
     // Useful when you're updating large (>22 field) POSOs.
@@ -290,6 +290,7 @@ object EnrichmentManager {
     event.refr_urlfragment = CU.truncate(event.refr_urlfragment, 255)
     event.refr_term = CU.truncate(event.refr_term, 255)
     event.se_label = CU.truncate(event.se_label, 255)
+    event.etl_tstamp = etlTstamp
 
     // Collect our errors on Failure, or return our event on Success 
     (useragent.toValidationNel |@| client.toValidationNel |@| pageUri.toValidationNel |@| geoLocation.toValidationNel |@| refererUri.toValidationNel |@| transform |@| campaign) {

3-enrich/scala-hadoop-enrich/project/Dependencies.scala

@@ -33,7 +33,7 @@ object Dependencies {
     val scalding         = "0.8.11"
     val scalaz7          = "7.0.0"
     val snowplowRawEvent = "0.1.0"
-    val commonEnrich     = "0.4.0"
+    val commonEnrich     = "0.5.0-SNAPSHOT"
     // Scala (test only)
     val specs2           = "1.14"
     val scalazSpecs2     = "0.1.2"

3-enrich/scala-hadoop-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich.hadoop/EtlJob.scala

@@ -60,9 +60,9 @@ object EtlJob {
    *         flatMap, will include any validation errors
    *         contained within the ValidatedMaybeCanonicalInput
    */
-  def toCanonicalOutput(geo: IpGeo, anonOctets: AnonOctets, input: ValidatedMaybeCanonicalInput): ValidatedMaybeCanonicalOutput = {
+  def toCanonicalOutput(geo: IpGeo, anonOctets: AnonOctets, input: ValidatedMaybeCanonicalInput, etlTstamp: String): ValidatedMaybeCanonicalOutput = {
     input.flatMap {
-      _.cata(EnrichmentManager.enrichEvent(geo, etlVersion, anonOctets, _).map(_.some),
+      _.cata(EnrichmentManager.enrichEvent(geo, etlVersion, anonOctets, etlTstamp, _).map(_.some),
              none.success)
     }
   }
@@ -140,6 +140,8 @@ class EtlJob(args: Args) extends Job(args) {
   val ipGeoFile = EtlJob.installIpGeoFile(etlConfig.maxmindFile)
   lazy val ipGeo = EtlJob.createIpGeo(ipGeoFile)
 
+  val etlTstamp = etlConfig.etl_tstamp
+
   // Aliases for our job
   val input = MultipleTextLineFiles(etlConfig.inFolder).read
   val goodOutput = Tsv(etlConfig.outFolder)
@@ -155,7 +157,7 @@ class EtlJob(args: Args) extends Job(args) {
   // TODO: let's fix this Any typing
   val common = trappableInput
     .map('line -> 'output) { l: Any =>
-      EtlJob.toCanonicalOutput(ipGeo, etlConfig.anonOctets, loader.toCanonicalInput(l))
+      EtlJob.toCanonicalOutput(ipGeo, etlConfig.anonOctets, loader.toCanonicalInput(l), etlTstamp)
     }
 
   // Handle bad rows

3-enrich/scala-hadoop-enrich/src/main/scala/com.snowplowanalytics.snowplow.enrich.hadoop/EtlJobConfig.scala

@@ -28,6 +28,7 @@ import com.twitter.scalding.Args
 import common.utils.ConversionUtils
 import common.enrichments.PrivacyEnrichments.AnonOctets
 import AnonOctets._
+import com.snowplowanalytics.snowplow.enrich.common.enrichments.EventEnrichments
 
 // This project
 import utils.ScalazArgs
@@ -111,7 +112,7 @@ 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 etlTstamp = args.requiredz("etl_tstamp").flatMap(t => EventEnrichments.extractTimestamp("etl_tstamp", t))
     val exceptionsFolder = args.optionalz("exceptions_folder")
 
     (inFolder.toValidationNel |@| inFormat.toValidationNel |@| maxmindFile.toValidationNel |@| outFolder.toValidationNel |@| badFolder.toValidationNel |@| anonOctets.toValidationNel |@| etlTstamp.toValidationNel |@| exceptionsFolder.toValidationNel) { EtlJobConfig(_,_,_,_,_,_,_,_) }

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/JobSpecHelpers.scala

@@ -23,6 +23,10 @@ import org.specs2.matcher.Matchers._
 // Scalding
 import com.twitter.scalding._
 
+// Joda-Time
+import org.joda.time.{DateTime, DateTimeZone}
+import org.joda.time.format.DateTimeFormat
+
 // Snowplow Common Enrich
 import common.outputs.CanonicalOutput
 
@@ -35,7 +39,9 @@ object JobSpecHelpers {
   /**
    * The current version of our Hadoop ETL
    */
-  val EtlVersion = "hadoop-0.5.0-common-0.4.0"
+  val EtlVersion = "hadoop-0.5.0-common-0.5.0-SNAPSHOT"
+
+  val EtlTimestamp = "2001-09-09 01:46:40.000"
 
   /**
    * Fields in our CanonicalOutput which are unmatchable
@@ -146,5 +152,6 @@ object JobSpecHelpers {
       arg("output_folder", "outputFolder").
       arg("bad_rows_folder", "badFolder").
       arg("anon_ip_octets", anonOctets).
-      arg("exceptions_folder", "exceptionsFolder")
+      arg("exceptions_folder", "exceptionsFolder").
+      arg("etl_tstamp", "1000000000000")
 }

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/Apr2014CfLineSpec.scala

@@ -43,6 +43,7 @@ object Apr2014CfLineSpec {
   val expected = List(
     "snowplowweb",
     "web",
+    EtlTimestamp,
     "2014-04-29 09:00:54.000",
     "2014-04-29 09:00:54.889",
     "page_ping",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/Aug2013CfLineSpec.scala

@@ -43,6 +43,7 @@ object Aug2013CfLineSpec {
   val expected = List(
     "snowplowweb",
     "web",
+    EtlTimestamp,
     "2013-08-29 00:18:48.000",
     "2013-08-29 00:19:17.970",
     "page_view",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/CljTomcatLineSpec.scala

@@ -42,6 +42,7 @@ object CljTomcatLineSpec {
   val expected = List(
     "snowplowweb",
     "web",
+    EtlTimestamp,
     "2013-10-07 19:47:54.000",
     "2013-10-07 19:47:54.123",
     "page_view",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/FutureCfLineSpec.scala

@@ -43,6 +43,7 @@ object FutureCfLineSpec {
   val expected = List(
     "snowplowweb",
     "web",
+    EtlTimestamp,
     "2015-04-29 09:00:54.000",
     "2015-04-29 08:00:54.889",
     "page_ping",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/LateOct2013CfLineSpec.scala

@@ -43,6 +43,7 @@ object LateOct2013CfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2013-10-22 00:41:30.000",
     "2013-10-22 00:41:53.725",
     "page_view",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/Oct2013CfLineSpec.scala

@@ -43,6 +43,7 @@ object Oct2013CfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2013-10-07 23:35:30.000",
     "2013-10-07 23:35:27.571",
     "page_ping",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/PagePingCfLineSpec.scala

@@ -42,6 +42,7 @@ object PagePingCfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2013-03-25 02:04:00.000",
     "2013-03-25 02:03:37.342",
     "page_ping",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/PageViewCfLineSpec.scala

@@ -42,6 +42,7 @@ object PageViewCfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2012-05-24 00:06:42.000",
     "2013-03-25 13:52:09.188",
     "page_view",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/Sep2013CfLineSpec.scala

@@ -43,6 +43,7 @@ object Sep2013CfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2013-10-07 21:32:22.000",
     "2013-10-07 21:30:37.923",
     "page_view",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/StructEventCfLineSpec.scala

@@ -42,6 +42,7 @@ object StructEventCfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2012-05-27 11:35:53.000",
     "2013-03-25 17:02:49.450",
     "struct",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/TransactionCfLineSpec.scala

@@ -42,6 +42,7 @@ object TransactionCfLineSpec {
   val expected = List(
     "CFe23a",
     null, // Not set (legacy input line)
+    EtlTimestamp,
     "2012-05-27 11:35:53.000",
     "2013-03-25 02:03:37.342",
     "transaction",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/TransactionItemCfLineSpec.scala

@@ -42,6 +42,7 @@ object TransactionItemCfLineSpec {
   val expected = List(
     "CFe23a",
     null, // Not set (legacy input line)
+    EtlTimestamp,
     "2012-05-27 11:35:53.000",
     "2013-03-25 02:03:37.342",
     "transaction_item",

3-enrich/scala-hadoop-enrich/src/test/scala/com.snowplowanalytics.snowplow.enrich.hadoop/good/UnstructEventCfLineSpec.scala

@@ -42,6 +42,7 @@ object UnstructEventCfLineSpec {
   val expected = List(
     "pbzsite",
     "web",
+    EtlTimestamp,
     "2012-05-27 11:35:53.000",
     "2013-03-25 17:02:49.450",
     "unstruct",

CHANGELOG

@@ -1,5 +1,6 @@
 Version 0.9.X (2014-XX-XX)
 --------------------------
+Scala Common Enrich: stored etl_tstamp in new field in CanonicalOutput (#818)
 Redshift: bumped table-def to 0.4.0
 Redshift: added etl_tstamp to atomic.events (#819)
 Postgres: bumped table-def to 0.3.0