diff options
Diffstat (limited to 'scalding/src/main/scala')
-rw-r--r-- | scalding/src/main/scala/sandcrawler/MissingColumnDumpJob.scala | 59 |
1 files changed, 59 insertions, 0 deletions
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) + } +} |