aboutsummaryrefslogtreecommitdiffstats
path: root/scalding/src/main/scala
diff options
context:
space:
mode:
authorBryan Newbold <bnewbold@archive.org>2018-05-23 12:27:59 -0700
committerBryan Newbold <bnewbold@archive.org>2018-05-24 00:02:36 -0700
commit4ba428db30593b67283dd90b92141f99840dc78e (patch)
treef63c8e146e7f90a530abfebdb993ab45d57426d5 /scalding/src/main/scala
parent29e4a83ff76da07bc6ad5d3f49d746ee0bc72023 (diff)
downloadsandcrawler-4ba428db30593b67283dd90b92141f99840dc78e.tar.gz
sandcrawler-4ba428db30593b67283dd90b92141f99840dc78e.zip
rename jvm/scalding directories
Diffstat (limited to 'scalding/src/main/scala')
-rw-r--r--scalding/src/main/scala/example/SimpleHBaseSourceExample.scala36
-rw-r--r--scalding/src/main/scala/example/WordCountJob.scala12
-rw-r--r--scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala41
3 files changed, 89 insertions, 0 deletions
diff --git a/scalding/src/main/scala/example/SimpleHBaseSourceExample.scala b/scalding/src/main/scala/example/SimpleHBaseSourceExample.scala
new file mode 100644
index 0000000..fe2a120
--- /dev/null
+++ b/scalding/src/main/scala/example/SimpleHBaseSourceExample.scala
@@ -0,0 +1,36 @@
+package example
+
+import com.twitter.scalding.{Tsv, Args}
+import parallelai.spyglass.base.JobBase
+import org.apache.log4j.{Level, Logger}
+import parallelai.spyglass.hbase.{HBasePipeConversions, HBaseSource}
+import parallelai.spyglass.hbase.HBaseConstants.SourceMode
+import cascading.tuple.Fields
+import cascading.property.AppProps
+import java.util.Properties
+
+/**
+ * Simple example of HBaseSource usage
+ */
+class SimpleHBaseSourceExample(args: Args) extends JobBase(args) with HBasePipeConversions {
+
+ val isDebug: Boolean = args("debug").toBoolean
+
+ if (isDebug) Logger.getRootLogger.setLevel(Level.DEBUG)
+
+ val output = args("output")
+
+ val hbs = new HBaseSource(
+ "table_name",
+ //"quorum_name:2181",
+ "mtrcs-zk1.us.archive.org:2181", // HBase Zookeeper server (to get runtime config info; can be array?)
+ new Fields("key"),
+ List("column_family"),
+ List(new Fields("column_name1", "column_name2")),
+ sourceMode = SourceMode.GET_LIST, keyList = List("1", "2", "3"))
+ .read
+ .debug
+ .fromBytesWritable(new Fields("key", "column_name1", "column_name2"))
+ .write(Tsv(output format "get_list"))
+
+ }
diff --git a/scalding/src/main/scala/example/WordCountJob.scala b/scalding/src/main/scala/example/WordCountJob.scala
new file mode 100644
index 0000000..0e63fed
--- /dev/null
+++ b/scalding/src/main/scala/example/WordCountJob.scala
@@ -0,0 +1,12 @@
+package example
+
+import com.twitter.scalding._
+
+class WordCountJob(args: Args) extends Job(args) {
+ TypedPipe.from(TextLine(args("input")))
+ .flatMap { line => line.split("\\s+") }
+ .map { word => (word, 1L) }
+ .sumByKey
+ // The compiler will enforce the type coming out of the sumByKey is the same as the type we have for our sink
+ .write(TypedTsv[(String, Long)](args("output")))
+}
diff --git a/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala b/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala
new file mode 100644
index 0000000..5df6b2e
--- /dev/null
+++ b/scalding/src/main/scala/sandcrawler/HBaseRowCountJob.scala
@@ -0,0 +1,41 @@
+package sandcrawler
+
+import com.twitter.scalding._
+import parallelai.spyglass.base.JobBase
+import parallelai.spyglass.hbase.{HBaseSource, HBasePipeConversions, HBaseConstants}
+import parallelai.spyglass.hbase.HBaseConstants.SourceMode
+import cascading.tuple.Fields
+import cascading.property.AppProps
+import java.util.Properties
+
+
+class HBaseRowCountJob(args: Args) extends JobBase(args) with HBasePipeConversions {
+
+
+ // For now doesn't actually count, just dumps a "word count"
+
+ val output = args("output")
+
+ val hbs = new HBaseSource(
+ //"table_name",
+ //"quorum_name:2181",
+ "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?)
+ new Fields("key"),
+ List("file"),
+ List(new Fields("size", "mimetype")),
+ sourceMode = SourceMode.GET_LIST, keyList = List("sha1:K2DKSSVTXWPRMFDTWSTCQW3RVWRIOV3Q", "sha1:C3YNNEGH5WAG5ZAAXWAEBNXJWT6CZ3WU"))
+ .read
+ .debug
+ .fromBytesWritable(new Fields("key"))
+ .write(Tsv(output format "get_list"))
+
+ /*
+ List("column_family"),
+ sourceMode = SourceMode.SCAN_ALL)
+ .read
+ .debug
+ .fromBytesWritable(new Fields("key"))
+ .write(Tsv(output format "get_list"))
+ */
+}