diff options
| author | Bryan Newbold <bnewbold@archive.org> | 2018-07-13 18:42:33 -0400 | 
|---|---|---|
| committer | Bryan Newbold <bnewbold@archive.org> | 2018-07-13 18:42:33 -0400 | 
| commit | f955fef045d3c78e351a639546a27ab0a53fc0aa (patch) | |
| tree | 5d25aa7e44d8bc196ac20f2e3a187bd0798ce134 /scalding | |
| parent | 0e74b5f49950c5c46d9be5e99066bee71cc8828f (diff) | |
| parent | 071f4ff633eaa2669c4afcfc421f95cdc0a6765f (diff) | |
| download | sandcrawler-f955fef045d3c78e351a639546a27ab0a53fc0aa.tar.gz sandcrawler-f955fef045d3c78e351a639546a27ab0a53fc0aa.zip | |
Merge branch 'bnewbold-args'
Diffstat (limited to 'scalding')
5 files changed, 30 insertions, 16 deletions
| diff --git a/scalding/src/main/scala/sandcrawler/HBaseCountJob.scala b/scalding/src/main/scala/sandcrawler/HBaseCountJob.scala index 815478b..863c16f 100644 --- a/scalding/src/main/scala/sandcrawler/HBaseCountJob.scala +++ b/scalding/src/main/scala/sandcrawler/HBaseCountJob.scala @@ -13,7 +13,9 @@ class HBaseCountJob(args: Args, colSpec: String) extends JobBase(args) with HBas    HBaseBuilder.parseColSpec(colSpec)    val Col: String = colSpec.split(":")(1) -  HBaseCountJob.getHBaseSource(colSpec) +  HBaseCountJob.getHBaseSource(args("hbase-table"), +                               args("zookeeper-hosts"), +                               colSpec)      .read      .fromBytesWritable(Symbol(Col))      .debug @@ -22,9 +24,9 @@ class HBaseCountJob(args: Args, colSpec: String) extends JobBase(args) with HBas  }  object HBaseCountJob { -  def getHBaseSource(colSpec: String) = HBaseBuilder.build( -    "wbgrp-journal-extract-0-qa",     // HBase Table Name -    "mtrcs-zk1.us.archive.org:2181",  // HBase Zookeeper server (to get runtime config info; can be array?) +  def getHBaseSource(hbaseTable: String, zookeeperHosts: String, colSpec: String) = HBaseBuilder.build( +    hbaseTable,      // HBase Table Name +    zookeeperHosts,  // HBase Zookeeper server (to get runtime config info; can be array?)      List(colSpec),      SourceMode.SCAN_ALL)  } diff --git a/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala b/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala index 0c8e615..334b7a6 100644 --- a/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala +++ b/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala @@ -10,12 +10,10 @@ import parallelai.spyglass.hbase.HBaseConstants.SourceMode  class HBaseRowCountJob(args: Args) extends JobBase(args) with HBasePipeConversions { - -  // For now doesn't actually count, just dumps a "word count" -    val output = args("output") -  HBaseRowCountJob.getHBaseSource +  HBaseRowCountJob.getHBaseSource(args("hbase-table"), +                                  args("zookeeper-hosts"))      .read      .debug      .groupAll { _.size('count) } @@ -23,9 +21,14 @@ class HBaseRowCountJob(args: Args) extends JobBase(args) with HBasePipeConversio  }  object HBaseRowCountJob { -  def getHBaseSource = HBaseBuilder.build( -    "wbgrp-journal-extract-0-qa",     // HBase Table Name -    "mtrcs-zk1.us.archive.org:2181",  // HBase Zookeeper server (to get runtime config info; can be array?) -    List("file:size", "file:mime"), -    SourceMode.SCAN_ALL) + +  // eg, "wbgrp-journal-extract-0-qa", "mtrcs-zk1.us.archive.org:2181" +  def getHBaseSource(hbaseTable: String, zookeeperHosts: String) : HBaseSource = { +    return HBaseBuilder.build( +      hbaseTable, +      zookeeperHosts, +      List("file:size"), +      SourceMode.SCAN_ALL) +  } +  } diff --git a/scalding/src/test/scala/sandcrawler/HBaseMimeCountTest.scala b/scalding/src/test/scala/sandcrawler/HBaseMimeCountTest.scala index cc400c5..fde2290 100644 --- a/scalding/src/test/scala/sandcrawler/HBaseMimeCountTest.scala +++ b/scalding/src/test/scala/sandcrawler/HBaseMimeCountTest.scala @@ -16,6 +16,7 @@ import scala._  class HBaseMimeCountTest extends FunSpec with TupleConversions {    val output = "/tmp/testOutput" +  val (testTable, testHost) = ("test-table", "dummy-host:2181")    val log = LoggerFactory.getLogger(this.getClass.getName) @@ -40,8 +41,10 @@ class HBaseMimeCountTest extends FunSpec with TupleConversions {      .arg("test", "")      .arg("app.conf.path", "app.conf")      .arg("output", output) +    .arg("hbase-table", testTable) +    .arg("zookeeper-hosts", testHost)      .arg("debug", "true") -    .source[Tuple](HBaseCountJob.getHBaseSource("file:mime"), +    .source[Tuple](HBaseCountJob.getHBaseSource(testTable, testHost, "file:mime"),        sampleData.map(l => new Tuple(l.map(s => {new ImmutableBytesWritable(Bytes.toBytes(s))}):_*)))        .sink[Tuple](Tsv(output)) {          outputBuffer => diff --git a/scalding/src/test/scala/sandcrawler/HBaseRowCountTest.scala b/scalding/src/test/scala/sandcrawler/HBaseRowCountTest.scala index 6f61eb3..3424a36 100644 --- a/scalding/src/test/scala/sandcrawler/HBaseRowCountTest.scala +++ b/scalding/src/test/scala/sandcrawler/HBaseRowCountTest.scala @@ -19,6 +19,7 @@ import scala._  class HBaseRowCountTest extends FunSpec with TupleConversions {    val output = "/tmp/testOutput" +  val (testTable, testHost) = ("test-table", "dummy-host:2181")    val log = LoggerFactory.getLogger(this.getClass.getName) @@ -37,8 +38,10 @@ class HBaseRowCountTest extends FunSpec with TupleConversions {      .arg("test", "")      .arg("app.conf.path", "app.conf")      .arg("output", output) +    .arg("hbase-table", testTable) +    .arg("zookeeper-hosts", testHost)      .arg("debug", "true") -    .source[Tuple](HBaseRowCountJob.getHBaseSource, +    .source[Tuple](HBaseRowCountJob.getHBaseSource(testTable, testHost),        sampleData.map(l => new Tuple(l.map(s => {new ImmutableBytesWritable(Bytes.toBytes(s))}):_*)))        .sink[Tuple](Tsv(output)) {        outputBuffer => diff --git a/scalding/src/test/scala/sandcrawler/HBaseStatusCountTest.scala b/scalding/src/test/scala/sandcrawler/HBaseStatusCountTest.scala index 056347c..fca9c69 100644 --- a/scalding/src/test/scala/sandcrawler/HBaseStatusCountTest.scala +++ b/scalding/src/test/scala/sandcrawler/HBaseStatusCountTest.scala @@ -16,6 +16,7 @@ import scala._  class HBaseStatusCountTest extends FunSpec with TupleConversions {    val output = "/tmp/testOutput" +  val (testTable, testHost) = ("test-table", "dummy-host:2181")    val log = LoggerFactory.getLogger(this.getClass.getName) @@ -40,8 +41,10 @@ class HBaseStatusCountTest extends FunSpec with TupleConversions {      .arg("test", "")      .arg("app.conf.path", "app.conf")      .arg("output", output) +    .arg("hbase-table", testTable) +    .arg("zookeeper-hosts", testHost)      .arg("debug", "true") -    .source[Tuple](HBaseCountJob.getHBaseSource("grobid0:status_code"), +    .source[Tuple](HBaseCountJob.getHBaseSource(testTable, testHost, "grobid0:status_code"),        sampleData.map(l => new Tuple(l.map(s => {new ImmutableBytesWritable(Bytes.toBytes(s))}):_*)))        .sink[Tuple](Tsv(output)) {          outputBuffer => | 
