From 24b35b230a0e2b45fd9607b0bf9f6090c4bfab52 Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 21 Aug 2018 14:36:50 -0700 Subject: WIP: MissingColumnDumpJob This version won't actually work because SpyGlass doesn't actually support SCAN filters (among other things) --- .../scala/sandcrawler/MissingColumnDumpJob.scala | 59 ++++++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100644 scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala diff --git a/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala b/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala new file mode 100644 index 0000000..de0228d --- /dev/null +++ b/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala @@ -0,0 +1,59 @@ +package sandcrawler + +import java.util.Properties + +import cascading.property.AppProps +import cascading.tuple.Fields +import com.twitter.scalding._ +import org.apache.hadoop.hbase.util.Bytes +import org.apache.hadoop.hbase.client.Scan +import org.apache.hadoop.hbase.filter.SingleColumnValueFilter +import org.apache.hadoop.hbase.filter.CompareFilter +import parallelai.spyglass.base.JobBase +import parallelai.spyglass.hbase.HBaseConstants.SourceMode +import parallelai.spyglass.hbase.HBasePipeConversions +import parallelai.spyglass.hbase.HBaseRawSource +import parallelai.spyglass.hbase.HBaseSource + +class MissingColumnDumpJob(args: Args) extends JobBase(args) with HBasePipeConversions { + + val output = args("output") + + MissingColumnDumpJob.getHBaseSource( + args("hbase-table"), + args("zookeeper-hosts"), + args("column")) + .read + .fromBytesWritable('key) + .write(Tsv(output)) +} + +object MissingColumnDumpJob { + + // eg, "wbgrp-journal-extract-0-qa",7 "mtrcs-zk1.us.archive.org:2181" + def getHBaseSource(hbaseTable: String, zookeeperHosts: String, col: String) : HBaseSource = { + + val colFamily = col.split(":")(0) + val colColumn = col.split(":")(1) + val scan = new Scan + val filter = new SingleColumnValueFilter( + Bytes.toBytes(colFamily), + Bytes.toBytes(colColumn), + CompareFilter.CompareOp.EQUAL, + Bytes.toBytes("") + ) + filter.setFilterIfMissing(false) + scan.setFilter(filter) + val scanner = HBaseRawSource.convertScanToString(scan) + val (families, fields) = HBaseBuilder.parseColSpecs(List("f:c", col)) + + new HBaseRawSource( + hbaseTable, + zookeeperHosts, + new Fields("key"), + families, + fields, + SourceMode.SCAN_ALL, + base64Scan = scanner) + } +} -- cgit v1.2.3 From 01dcfcb9786889228b8f83d211351202bf3cf4fb Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 21 Aug 2018 17:11:40 -0700 Subject: rewrite MissingColumnDumpJob as a join (sigh) --- .../scala/sandcrawler/MissingColumnDumpJob.scala | 66 ++++++++++++---------- 1 file changed, 37 insertions(+), 29 deletions(-) diff --git a/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala b/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala index de0228d..cc3bf23 100644 --- a/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala +++ b/scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala @@ -5,55 +5,63 @@ import java.util.Properties import cascading.property.AppProps import cascading.tuple.Fields import com.twitter.scalding._ -import org.apache.hadoop.hbase.util.Bytes -import org.apache.hadoop.hbase.client.Scan -import org.apache.hadoop.hbase.filter.SingleColumnValueFilter -import org.apache.hadoop.hbase.filter.CompareFilter +import com.twitter.scalding.typed.TDsl._ import parallelai.spyglass.base.JobBase import parallelai.spyglass.hbase.HBaseConstants.SourceMode import parallelai.spyglass.hbase.HBasePipeConversions -import parallelai.spyglass.hbase.HBaseRawSource import parallelai.spyglass.hbase.HBaseSource +// This nasty, no-good, horrible Job outputs a list of keys ("sha1:A234...") +// for which the given "column" does not have a value set. +// It does this using a self-join because SpyGlass's HBase SCAN support seems +// to be extremely limited. class MissingColumnDumpJob(args: Args) extends JobBase(args) with HBasePipeConversions { val output = args("output") - MissingColumnDumpJob.getHBaseSource( + val allKeys : TypedPipe[String] = MissingColumnDumpJob.getHBaseKeySource( + args("hbase-table"), + args("zookeeper-hosts")) + .read + .fromBytesWritable('key) + .toTypedPipe[String]('key) + + val existingKeys : TypedPipe[(String,Boolean)] = MissingColumnDumpJob.getHBaseColSource( args("hbase-table"), args("zookeeper-hosts"), args("column")) .read .fromBytesWritable('key) - .write(Tsv(output)) + .toTypedPipe[String]('key) + .map{ key => (key, true) } + + val missingKeys : TypedPipe[String] = allKeys + .groupBy( identity ) + .leftJoin(existingKeys.groupBy(_._1)) + .toTypedPipe + .collect { case (key, (_, None)) => key } + + missingKeys + .write(TypedTsv[String](output)) + } object MissingColumnDumpJob { // eg, "wbgrp-journal-extract-0-qa",7 "mtrcs-zk1.us.archive.org:2181" - def getHBaseSource(hbaseTable: String, zookeeperHosts: String, col: String) : HBaseSource = { - - val colFamily = col.split(":")(0) - val colColumn = col.split(":")(1) - val scan = new Scan - val filter = new SingleColumnValueFilter( - Bytes.toBytes(colFamily), - Bytes.toBytes(colColumn), - CompareFilter.CompareOp.EQUAL, - Bytes.toBytes("") - ) - filter.setFilterIfMissing(false) - scan.setFilter(filter) - val scanner = HBaseRawSource.convertScanToString(scan) - val (families, fields) = HBaseBuilder.parseColSpecs(List("f:c", col)) - - new HBaseRawSource( + def getHBaseColSource(hbaseTable: String, zookeeperHosts: String, col: String) : HBaseSource = { + HBaseBuilder.build( + hbaseTable, + zookeeperHosts, + List(col), + SourceMode.SCAN_ALL) + } + + def getHBaseKeySource(hbaseTable: String, zookeeperHosts: String) : HBaseSource = { + HBaseBuilder.build( hbaseTable, zookeeperHosts, - new Fields("key"), - families, - fields, - SourceMode.SCAN_ALL, - base64Scan = scanner) + List("f:c"), + SourceMode.SCAN_ALL) } } -- cgit v1.2.3 From bd9c6564a07eb1b5d9c6a25942a198dc3800281b Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Tue, 21 Aug 2018 17:14:25 -0700 Subject: add please for keysmissingcolumn --- please | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/please b/please index b32dd79..a5e7f15 100755 --- a/please +++ b/please @@ -215,6 +215,30 @@ def run_colcount(args): env=args.env) subprocess.call(cmd, shell=True) +def run_keysmissingcol(args): + if args.rebuild: + rebuild_scalding() + print("Starting keysmissingcol job...") + output = "{}/output-{}/{}-keysmissingcol-{}".format( + HDFS_DIR, + args.env, + datetime.strftime(datetime.now(), "%Y-%m-%d-%H%M.%S"), + args.column.replace(":", "_")) + cmd = """hadoop jar \ + scalding/target/scala-2.11/sandcrawler-assembly-0.2.0-SNAPSHOT.jar \ + com.twitter.scalding.Tool sandcrawler.MissingColDumpJob \ + --hdfs \ + --app.conf.path scalding/ia_cluster.conf \ + --hbase-table wbgrp-journal-extract-0-{env} \ + --zookeeper-hosts {zookeeper_hosts} \ + --column {column} \ + --output {output}""".format( + output=output, + zookeeper_hosts=ZOOKEEPER_HOSTS, + column=args.column, + env=args.env) + subprocess.call(cmd, shell=True) + def main(): parser = argparse.ArgumentParser() @@ -264,6 +288,11 @@ def main(): sub_colcount.add_argument('column', help="column name to use in count") + sub_keysmissingcol = subparsers.add_parser('keys-missing-col') + sub_keysmissingcol.set_defaults(func=run_matchcrossref) + sub_keysmissingcol.add_argument('column', + help="column to SCAN for missing keys") + args = parser.parse_args() if not args.__dict__.get("func"): print("tell me what to do! (try --help)") -- cgit v1.2.3 From 9557323addb8ab0b7f84dc404e8ef3d76243f13f Mon Sep 17 00:00:00 2001 From: Bryan Newbold Date: Wed, 22 Aug 2018 03:02:06 +0000 Subject: fixes to please keys-missing-col --- please | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/please b/please index a5e7f15..023eeac 100755 --- a/please +++ b/please @@ -226,7 +226,7 @@ def run_keysmissingcol(args): args.column.replace(":", "_")) cmd = """hadoop jar \ scalding/target/scala-2.11/sandcrawler-assembly-0.2.0-SNAPSHOT.jar \ - com.twitter.scalding.Tool sandcrawler.MissingColDumpJob \ + com.twitter.scalding.Tool sandcrawler.MissingColumnDumpJob \ --hdfs \ --app.conf.path scalding/ia_cluster.conf \ --hbase-table wbgrp-journal-extract-0-{env} \ @@ -289,7 +289,7 @@ def main(): help="column name to use in count") sub_keysmissingcol = subparsers.add_parser('keys-missing-col') - sub_keysmissingcol.set_defaults(func=run_matchcrossref) + sub_keysmissingcol.set_defaults(func=run_keysmissingcol) sub_keysmissingcol.add_argument('column', help="column to SCAN for missing keys") -- cgit v1.2.3