diff options
| -rw-r--r-- | scalding/src/main/scala/sandcrawler/DumpGrobidXmlJob.scala | 41 |
1 files changed, 41 insertions, 0 deletions
diff --git a/scalding/src/main/scala/sandcrawler/DumpGrobidXmlJob.scala b/scalding/src/main/scala/sandcrawler/DumpGrobidXmlJob.scala new file mode 100644 index 0000000..1b178eb --- /dev/null +++ b/scalding/src/main/scala/sandcrawler/DumpGrobidXmlJob.scala @@ -0,0 +1,41 @@ +package sandcrawler + +import java.util.Properties + +import cascading.property.AppProps +import cascading.tuple.Fields +import com.twitter.scalding._ +import com.twitter.scalding.typed.TDsl._ +import org.apache.hadoop.hbase.io.ImmutableBytesWritable +import org.apache.hadoop.hbase.util.Bytes +import parallelai.spyglass.base.JobBase +import parallelai.spyglass.hbase.HBaseConstants.SourceMode +import parallelai.spyglass.hbase.HBasePipeConversions +import parallelai.spyglass.hbase.HBaseSource +import scala.util.parsing.json.JSONObject + +// Dumps the SHA1 key and grobid0:tei_xml columns, as TSV/JSON (two TSV +// columns: one is key, second is JSON). Used for partner delivery/sharing +class DumpGrobidXmlJob(args: Args) extends JobBase(args) with HBasePipeConversions { + + val metaPipe : TypedPipe[(String, String)] = HBaseBuilder.build(args("hbase-table"), + args("zookeeper-hosts"), + List("file:cdx", "grobid0:tei_xml"), + SourceMode.SCAN_ALL) + .read + .toTypedPipe[(ImmutableBytesWritable,ImmutableBytesWritable,ImmutableBytesWritable)](new Fields("key", "cdx", "tei_xml")) + .filter { case (_, cdx, tei_xml) => cdx != null && tei_xml != null } + .map { case (key, cdx, tei_xml) => + (Bytes.toString(key.copyBytes()), + JSONObject( + Map( + "pdf_hash" -> Bytes.toString(key.copyBytes()), + "cdx_metadata" -> Bytes.toString(key.copyBytes()), + "tei_xml" -> Bytes.toString(key.copyBytes()) + )).toString + ) + }; + + metaPipe.write(TypedTsv[(String,String)](args("output"))) + +} |