Added JDBCTap support.

5 files changed, 390 insertions, 0 deletions

diff --git a/src/main/scala/parallelai/spyglass/jdbc/JDBCSource.scala b/src/main/scala/parallelai/spyglass/jdbc/JDBCSource.scala
new file mode 100644
index 0000000..2a08b7d
--- /dev/null
+++ b/src/main/scala/parallelai/spyglass/jdbc/JDBCSource.scala
@@ -0,0 +1,56 @@
+package parallelai.spyglass.jdbc
+
+import com.twitter.scalding.AccessMode
+import com.twitter.scalding.Hdfs
+import com.twitter.scalding.Mode
+import com.twitter.scalding.Read
+import com.twitter.scalding.Source
+import com.twitter.scalding.Write
+import cascading.scheme.Scheme
+import cascading.tap.Tap
+import cascading.tuple.Fields
+import org.apache.hadoop.mapred.RecordReader
+import org.apache.hadoop.mapred.OutputCollector
+import org.apache.hadoop.mapred.JobConf
+
+class JDBCSource(
+    tableName: String = "tableName",
+    driverName: String = "com.mysql.jdbc.Driver",
+    connectionString: String = "jdbc:mysql://<hostname>:<port>/<db_name>",
+    userId: String = "user",
+    password: String = "password",
+    columnNames: Array[String] = Array[String]("col1", "col2", "col3"),
+    columnDefs: Array[String] = Array[String]("data_type", "data_type", "data_type"),
+    primaryKeys: Array[String] = Array[String]("primary_key"),
+    fields: Fields = new Fields("fld1", "fld2", "fld3"),
+    orderBy: Array[String] = null,
+    updateBy: Array[String] = null,
+    updateByFields: Fields = null
+  ) extends Source {
+
+  override val hdfsScheme = new JDBCScheme(fields, columnNames, orderBy, updateByFields, updateBy)
+    .asInstanceOf[Scheme[JobConf, RecordReader[_, _], OutputCollector[_, _], _, _]]
+
+  override def createTap(readOrWrite: AccessMode)(implicit mode: Mode): Tap[_, _, _] = {
+    val jdbcScheme = hdfsScheme match {
+      case jdbc: JDBCScheme => jdbc
+      case _ => throw new ClassCastException("Failed casting from Scheme to JDBCScheme")
+    }
+    mode match {
+      case hdfsMode @ Hdfs(_, _) => readOrWrite match {
+        case Read => {
+          val tableDesc = new TableDesc(tableName, columnNames, columnDefs, primaryKeys)
+          val jdbcTap = new JDBCTap(connectionString, userId, password, driverName, tableDesc, jdbcScheme)
+          jdbcTap.asInstanceOf[Tap[_,_,_]]
+        }
+        case Write => {
+
+          val tableDesc = new TableDesc(tableName, columnNames, columnDefs, primaryKeys)
+          val jdbcTap = new JDBCTap(connectionString, userId, password, driverName, tableDesc, jdbcScheme)
+          jdbcTap.asInstanceOf[Tap[_,_,_]]
+        }
+      }
+      case _ => super.createTap(readOrWrite)(mode)
+    }
+  }
+}
diff --git a/src/main/scala/parallelai/spyglass/jdbc/testing/HdfsToJdbc.scala b/src/main/scala/parallelai/spyglass/jdbc/testing/HdfsToJdbc.scala
new file mode 100644
index 0000000..1544f47
--- /dev/null
+++ b/src/main/scala/parallelai/spyglass/jdbc/testing/HdfsToJdbc.scala
@@ -0,0 +1,57 @@
+package parallelai.spyglass.jdbc.testing
+
+import com.twitter.scalding.TextLine
+import com.twitter.scalding.Args
+import com.twitter.scalding.Tsv
+import com.twitter.scalding.mathematics.Matrix._
+import scala.math._
+import scala.math.BigDecimal.javaBigDecimal2bigDecimal
+import cascading.tuple.Fields
+import cascading.pipe.Pipe
+import com.twitter.scalding.Osv
+import parallelai.spyglass.base.JobBase
+import parallelai.spyglass.jdbc.JDBCSource
+
+class HdfsToJdbc (args: Args) extends JobBase(args) { 
+
+	implicit val implicitArgs: Args = args
+  
+    val scaldingInputPath = getString("input.scalding")  
+    log.info("Scalding sample input path => [" + scaldingInputPath + "]")
+  
+    val S_output = scaldingInputPath
+    val fileType = getString("fileType")
+    log.info("Input file type => " + fileType)
+  
+    val S_SCHEMA = List(
+		  'key_id,    'col1, 'col2, 'col3
+    )  
+    	
+	val url = "mysql01.prod.bigdata.bskyb.com"
+	val dbName = "skybet_db" 
+	val tableName = "skybet_hbase_betdetail_jdbc_test"
+	
+	  
+	val jdbcSource2 = new JDBCSource(
+		"db_name",
+		"com.mysql.jdbc.Driver",
+		"jdbc:mysql://<hostname>:<port>/<db_name>?zeroDateTimeBehavior=convertToNull",
+		"user",
+		"password",
+		Array[String]("KEY_ID", "COL1", "COL2", "COL3"),
+		Array[String]( "bigint(20)" ,  "varchar(45)"  ,  "varchar(45)"  ,  "bigint(20)"),
+		Array[String]("key_id"),
+		new Fields("key_id",    "col1",    "col2",    "col3")
+	)
+  
+    var piper:Pipe = null
+    if (fileType equals("Tsv"))
+    	piper = Tsv(S_output, S_SCHEMA).read
+    else
+    	piper = Osv(S_output, S_SCHEMA).read
+	   
+	val S_FLOW = 
+    	Tsv(S_output, S_SCHEMA).read
+    	.write(jdbcSource2)
+
+}
\ No newline at end of file
diff --git a/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWrite.scala b/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWrite.scala
new file mode 100644
index 0000000..30c03a2
--- /dev/null
+++ b/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWrite.scala
@@ -0,0 +1,200 @@
+package parallelai.spyglass.jdbc.testing
+
+import org.apache.log4j.Level
+import org.apache.log4j.LogManager
+import org.apache.log4j.Logger
+import com.twitter.scalding.Args
+import com.twitter.scalding.IterableSource
+import com.twitter.scalding.Tsv
+import cascading.pipe.Pipe
+import cascading.tuple.Fields
+import parallelai.spyglass.base.JobBase
+import parallelai.spyglass.jdbc.JDBCSource
+
+/**
+ * This integration-test expects some Jdbc table to exist
+ * with specific data - see GenerateTestingHTables.java
+ */
+
+// https://github.com/twitter/scalding/blob/develop/scalding-core/src/test/scala/com/twitter/scalding/BlockJoinTest.scala
+class JdbcSourceShouldReadWrite (args: Args) extends JobBase(args) {
+  
+  // Initiate logger
+  private val LOG: Logger = LogManager.getLogger(this.getClass)
+  
+  // Set to Level.DEBUG if --debug is passed in
+  val isDebug:Boolean = args.getOrElse("debug", "false").toBoolean
+  if (isDebug) { 
+    LOG.setLevel(Level.DEBUG)
+    LOG.info("Setting logging to Level.DEBUG")
+  }
+  
+  val url = "mysql01.prod.bigdata.bskyb.com"
+  val dbName = "skybet_db" 
+  val tableName = "skybet_hbase_betdetail_jdbc_test"
+
+  val jdbcSourceRead = new JDBCSource(
+	"TABLE_01",
+	"com.mysql.jdbc.Driver",
+	"jdbc:mysql://localhost:3306/sky_db?zeroDateTimeBehavior=convertToNull",
+	"root",
+	"password",
+	Array[String]("ID", "TEST_COLUMN1", "TEST_COLUMN2", "TEST_COLUMN3"),
+	Array[String]( "bigint(20)" ,  "varchar(45)"  ,  "varchar(45)"  ,    "bigint(20)"),
+	Array[String]("id"),
+	new Fields("key",    "column1",    "column2",    "column3"),
+	null, null, null
+  )
+  
+  val jdbcSourceWrite = new JDBCSource(
+	"TABLE_01",
+	"com.mysql.jdbc.Driver",
+	"jdbc:mysql://localhost:3306/sky_db?zeroDateTimeBehavior=convertToNull",
+	"root",
+	"password",
+	Array[String]("ID", "TEST_COLUMN1", "TEST_COLUMN2", "TEST_COLUMN3"),
+	Array[String]( "bigint(20)" ,  "varchar(45)"  ,  "varchar(45)"  ,    "bigint(20)"),
+	Array[String]("id"),
+	new Fields("key",    "column1",    "column2",    "column3"),
+	null, null, null
+  )  
+    
+  // -----------------------------
+  // ----- Tests for TABLE_01 ----
+  // -----------------------------
+  val TABLE_01_SCHEMA = List('key,'column1, 'column2, 'column3)
+  val tableName1 = "TABLE_01"
+
+  // -------------------- Test 01 --------------------
+  var testName01 = "Select_Test_Read_Count"
+  println("---- Running : " + testName01)
+  // Get everything from HBase testing table into a Pipe
+  val jdbc01 = jdbcSourceRead
+  		  	.read  	    
+  .groupAll { group => 
+  	   group.toList[String]('key -> 'key)
+  	   group.toList[String]('column1 -> 'column1)
+  	   group.toList[String]('column2 -> 'column2)
+  	   group.toList[String]('column3 -> 'column3)
+  }
+  .mapTo(('key, 'column1,  'column2,  'column3) -> 'jdbcdata) { x:(String,String,String,String) =>
+	     x._1 + " " + x._2 + " " + x._3 + " " + x._4
+  }
+  
+  // Calculate expected result for Test 01
+  var list01 = List(("1", "A", "X", "123"), ("2", "B", "Y", "234"),  ("3", "C", "Z", "345"))
+ 
+  // -------------------- Test 02 --------------------
+  val testName02 = "Select_Test_Read_Insert_Updated_Count"
+  println("---- Running : " + testName02)
+  
+  // Get everything from JDBC testing table into a Pipe
+  
+  val jdbcSourceReadUpdated = new JDBCSource(
+	"TABLE_02",
+	"com.mysql.jdbc.Driver",
+	"jdbc:mysql://localhost:3306/sky_db?zeroDateTimeBehavior=convertToNull",
+	"root",
+	"password",
+	Array[String]("ID", "TEST_COLUMN1", "TEST_COLUMN2", "TEST_COLUMN3"),
+	Array[String]( "bigint(20)" ,  "varchar(45)"  ,  "varchar(45)"  ,    "bigint(20)"),
+	Array[String]("id"),
+	new Fields("key",    "column1",    "column2",    "column3"),
+	null, null, null
+  )  
+  
+  val jdbc02 = jdbcSourceReadUpdated
+  .read  	    
+  .groupAll { group => 
+  	   group.toList[String]('key -> 'key)
+  	   group.toList[String]('column1 -> 'column1)
+  	   group.toList[String]('column2 -> 'column2)
+  	   group.toList[String]('column3 -> 'column3)
+  }
+  .mapTo(('key, 'column1,  'column2,  'column3) -> 'jdbcdata) { x:(String,String,String,String) =>
+	     x._1 + " " + x._2 + " " + x._3 + " " + x._4
+  }
+  
+  // Calculate expected result for Test 02
+  var list02 = List(("1", "A", "X", "123"), ("2", "B", "Y", "234"),  ("3", "C", "Z", "345"))
+  
+  // Store results of Scan Test 01
+  (
+    getTestResultPipe(getExpectedPipe(list01), jdbc01, testName01) ++
+    getTestResultPipe(getExpectedPipe(list02), jdbc02, testName02)
+  ).groupAll { group =>
+    group.sortBy('testName)
+  }
+  .write(Tsv("JdbcShouldRead"))
+  
+  
+  /**
+   * We assume the pipe is empty
+   * 
+   * We concatenate with a header - if the resulting size is 1 
+   * then the original size was 0 - then the pipe was empty :)
+   * 
+   * The result is then returned in a Pipe
+   */
+  def assertPipeIsEmpty ( jdbcPipe : Pipe , testName:String) : Pipe = { 
+    val headerPipe = IterableSource(List(testName), 'jdbcdata)
+    val concatenation = ( jdbcPipe ++ headerPipe ).groupAll{ group =>
+      group.size('size)
+    }
+    .project('size)
+    
+    val result = 
+      concatenation
+      .mapTo('size -> ('testName, 'result, 'expecteddata, 'jdbcdata)) { x:String => { 
+	      if (x == "1") { 
+	        (testName, "Success", "", "")
+	      } else {
+	        (testName, "Test Failed", "", "")
+	      }  
+    	}
+      }
+    
+    result
+  }
+  
+  /**
+   * Methods receives 2 pipes - and projects the results of testing
+   * 
+   * expectedPipe  should have a column 'expecteddata
+   * realJdbcPipe should have a column 'jdbcdata
+   */
+  def getTestResultPipe ( expectedPipe:Pipe , realJdbcPipe:Pipe, testName: String ): Pipe = {
+	val results = expectedPipe.insert('testName , testName)
+                 .joinWithTiny('testName -> 'testName, realJdbcPipe.insert('testName , testName))
+    .map(('expecteddata, 'jdbcdata)->'result) { x:(String,String) =>
+      	//println(x._1 + " === " + x._2)
+        if (x._1.equals(x._2)) 
+           "Success"
+        else
+           "Test Failed"
+    }
+    .project('testName, 'result, 'expecteddata, 'jdbcdata)
+    results
+  }
+
+  /** 
+   *  
+   */
+  def getExpectedPipe ( expectedList: List[(String,String,String,String)]) : Pipe = {
+    
+    val expectedPipe = 
+      IterableSource(expectedList, TABLE_01_SCHEMA)
+      .groupAll { group =>
+        group.toList[String]('key -> 'key)
+        group.toList[String]('column1 -> 'column1)
+        group.toList[String]('column2 -> 'column2)
+        group.toList[String]('column3 -> 'column3)
+        
+      }
+      .mapTo(('*) -> 'expecteddata) { x:(String,String,String,String) =>
+         x._1 + " " + x._2 + " " + x._3 + " " + x._4
+      }
+   expectedPipe
+  }
+  
+}
diff --git a/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWriteRunner.scala b/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWriteRunner.scala
new file mode 100644
index 0000000..f317834
--- /dev/null
+++ b/src/main/scala/parallelai/spyglass/jdbc/testing/JdbcSourceShouldReadWriteRunner.scala
@@ -0,0 +1,10 @@
+package parallelai.spyglass.jdbc.testing
+
+import parallelai.spyglass.base.JobRunner
+
+object JdbcSourceShouldReadWriteRunner extends App {
+  val appConfig = "/projects/applications.conf"
+  val libPath = "/*.jar"
+
+  JobRunner.main(Array(classOf[JdbcSourceShouldReadWrite].getName, "--hdfs", "--app.conf.path", appConfig, "--job.lib.path", libPath))
+}
\ No newline at end of file
diff --git a/src/main/scala/parallelai/spyglass/jdbc/testing/TablesComparison.scala b/src/main/scala/parallelai/spyglass/jdbc/testing/TablesComparison.scala
new file mode 100644
index 0000000..9fc09e4
--- /dev/null
+++ b/src/main/scala/parallelai/spyglass/jdbc/testing/TablesComparison.scala
@@ -0,0 +1,67 @@
+package parallelai.spyglass.jdbc.testing
+
+import com.twitter.scalding._
+import cascading.tuple.Fields
+import parallelai.spyglass.base.JobBase
+import parallelai.spyglass.jdbc.JDBCSource
+
+/**
+ * Compares whether two tables have the same data or not writing to HDFS the ids of the records that don't match.
+ * Now hardcoded for Skybet betdetail summation sample.
+ * To run it:
+ * bskyb.commons.scalding.base.JobRunner bskyb.commons.skybase.jdbc.testing.TablesComparison \
+ *    --app.conf.path /projects/application-hadoop.conf --hdfs  \
+ *    --job.lib.path file:///home/gfe01/IdeaProjects/commons/commons.hbase.skybase/alternateLocation
+ * @param args
+ */
+class TablesComparison(args: Args) extends JobBase(args) {
+
+  implicit val implicitArgs: Args = args
+  val conf = appConfig
+
+  val jdbcSink = new JDBCSource(
+    "table_name",
+    "com.mysql.jdbc.Driver",
+    "jdbc:mysql://<hostname>:<port>/<db_name>",
+    "skybet_user",
+    "zkb4Uo{C8",
+    Array[String]("BETLEG_ID", "CR_DATE", "EV_MKT_ID", "CUST_ID", "SET_DATETIME", "BET_DATETIME", "STATUS", "SOURCE", "BET_TYPE", "AFF_NAME", "CURRENCY_CODE", "BET_ID", "LEG_ID", "RECEIPT_NO", "STAKE", "REFUND", "WINNINGS", "PROFIT", "STAKE_GBP", "REFUND_GBP", "WINNINGS_GBP", "PROFIT_GBP", "NUM_SELS", "EXCH_RATE", "ACCT_NO", "BET_IP_ADDRESS", "NUM_DRAWS", "EXTRACT_DATE", "BET_TIME", "BET_DATE_TIME", "SET_DATE_TIME", "BET_DATE_MONTH", "SET_TIME_KEY", "BET_TIME_KEY", "SET_TIME", "SET_DATE", "BET_DATE", "PART_NO", "MARKET_SORT", "TAG", "PLACED_IN_RUNNING", "MAX_STAKE_SCALE", "ODDS_NUM", "ODDS_DEN", "EV_OC_ID", "USER_CLIENT_ID"),
+    Array[String]("bigint(20)", "varchar(45)", "varchar(45)", "bigint(20)", "varchar(45)", "varchar(45)", "char(1)", "varchar(45)", "char(5)", "varchar(45)", "char(3)", "bigint(20)", "bigint(20)", "varchar(24)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "decimal(12,2)", "smallint(6)", "decimal(12,2)", "varchar(45)", "char(15)", "int(11)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "varchar(45)", "int(11)", "varchar(45)", "varchar(45)", "char(1)", "double(10,5)", "int(11)", "int(11)", "bigint(20)", "varchar(45)"),
+    Array[String]("betleg_id"),
+    new Fields("betleg_id", "cr_date", "ev_mkt_id", "cust_id", "set_datetime", "bet_datetime", "status", "source", "bet_type", "aff_name", "currency_code", "bet_id", "leg_id", "receipt_no", "stake", "refund", "winnings", "profit", "stake_gbp", "refund_gbp", "winnings_gbp", "profit_gbp", "num_sels", "exch_rate", "acct_no", "bet_ip_address", "num_draws", "extract_date", "bet_time", "bet_date_time", "set_date_time", "bet_date_month", "set_time_key", "bet_time_key", "set_time", "set_date", "bet_date", "part_no", "market_sort", "tag", "placed_in_running", "max_stake_scale", "odds_num", "odds_den", "ev_oc_id", "user_client_id"),
+    Array[String]("BETLEG_ID"),
+    Array[String]("BETLEG_ID"),
+    new Fields("betleg_id")
+  )
+    .read
+    .insert('name, "betdetail")
+    .project('bet_id, 'part_no, 'leg_id)
+
+  //
+  // .write(new TextLine("testJDBCComparator/compare1"))
+
+  val jdbcSource2 = new JDBCSource(
+    "skybet_midas_bet_detail",
+    "com.mysql.jdbc.Driver",
+    "jdbc:mysql://mysql01.prod.bigdata.bskyb.com:3306/skybet_db?zeroDateTimeBehavior=convertToNull",
+    "skybet_user",
+    "zkb4Uo{C8",
+    Array[String]("BET_ID", "BET_TYPE_ID", "RECEIPT_NO", "NUM_SELS", "NUM_LINES", "BET_CHANNEL_CODE", "MOBILE_CLIENT_ID", "BET_AFFILIATE_ID", "BET_IP_ADDRESS", "LEG_ID", "LEG_TYPE", "OUTCOME_ID", "ACCT_ID", "BET_PLACED_DATETIME", "BET_PLACED_DATE", "BET_PLACED_TIME", "BET_SETTLED_DATETIME", "BET_SETTLED_DATE", "BET_SETTLED_TIME", "BET_STATUS", "STAKE", "REFUND", "'RETURN'", "PROFIT", "CURRENCY_TYPE_KEY", "EXCH_RATE", "STAKE_GBP", "REFUNDS_GBP", "RETURN_GBP", "PROFIT_GBP", "MARKET_TAG", "MARKET_SORT", "PLACED_IN_RUNNING", "ODDS_NUM", "ODDS_DEN", "BETLEG_ID", "PART_NO"),
+    Array[String]("bigint(20)", "varchar(16)", "varchar(32)", "int(10)", "int(10)", "char(1)", "varchar(32)", "varchar(32)", "varchar(15)", "int(11)", "char(1)", "bigint(20)", "bigint(20)", "datetime", "date", "time", "datetime", "date", "time", "varchar(32)", "decimal(8,3)", "decimal(8,3)", "decimal(8,3)", "decimal(8,3)", "varchar(32)", "decimal(8,3)", "decimal(8,3)", "decimal(8,3)", "decimal(8,3)", "decimal(8,3)", "varchar(32)", "varchar(32)", "varchar(32)", "int(11)", "int(11)"),
+    Array[String]("bet_id"),
+    new Fields("bet_id", "bet_type_id", "receipt_no", "num_sels", "num_lines", "bet_channel_code", "mobile_client_id", "bet_affiliate_id", "bet_ip_address", "leg_id", "leg_type", "outcome_id", "acct_id", "bet_placed_datetime", "bet_placed_date", "bet_placed_time", "bet_settled_datetime", "bet_settled_date", "bet_settled_time", "bet_status", "stake", "refund", "return", "profit", "currency_type_key", "exch_rate", "stake_gbp", "refunds_gbp", "return_gbp", "profit_gbp", "market_tag", "market_sort", "placed_in_running", "odds_num", "odds_den", "betleg_id", "part_no")
+
+  )
+    .read
+    .insert('name, "sample")
+    .project('bet_id, 'part_no, 'leg_id)
+
+  val uPipe = jdbcSink ++ jdbcSource2
+  uPipe
+    .groupBy('bet_id, 'part_no, 'leg_id) {
+    _.size
+  }.filter('size) {
+    x: Int => x != 2
+  }
+    .write(new TextLine("testJDBCComparator/result"))
+}
\ No newline at end of file