1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
package parallelai.spyglass.hbase
import cascading.pipe.Pipe
import cascading.pipe.assembly.Coerce
import cascading.scheme.Scheme
import cascading.tap.{ Tap, SinkMode }
import cascading.tuple.Fields
import org.apache.hadoop.mapred.{ RecordReader, OutputCollector, JobConf }
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.JavaConversions._
import scala.collection.mutable.WrappedArray
import com.twitter.scalding._
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil
import org.apache.hadoop.hbase.util.Base64
import java.io.ByteArrayOutputStream
import java.io.DataOutputStream
object HBaseRawSource {
/**
* Converts a scan object to a base64 string that can be passed to HBaseRawSource
* @param scan
* @return base64 string representation
*/
def convertScanToString(scan: Scan) = {
val out = new ByteArrayOutputStream();
val dos = new DataOutputStream(out);
scan.write(dos);
Base64.encodeBytes(out.toByteArray());
}
}
/**
* @author Rotem Hermon
*
* HBaseRawSource is a scalding source that passes the original row (Result) object to the
* mapper for customized processing.
*
* @param tableName The name of the HBase table to read
* @param quorumNames HBase quorum
* @param familyNames Column families to get (source, if null will get all) or update to (sink)
* @param writeNulls Should the sink write null values. default = true. If false, null columns will not be written
* @param base64Scan An optional base64 encoded scan object
* @param sinkMode If REPLACE the output table will be deleted before writing to
*
*/
class HBaseRawSource(
tableName: String,
quorumNames: String = "localhost",
familyNames: Array[String],
writeNulls: Boolean = true,
base64Scan: String = null,
sinkMode: SinkMode = null) extends Source {
override val hdfsScheme = new HBaseRawScheme(familyNames, writeNulls)
.asInstanceOf[Scheme[JobConf, RecordReader[_, _], OutputCollector[_, _], _, _]]
override def createTap(readOrWrite: AccessMode)(implicit mode: Mode): Tap[_, _, _] = {
val hBaseScheme = hdfsScheme match {
case hbase: HBaseRawScheme => hbase
case _ => throw new ClassCastException("Failed casting from Scheme to HBaseRawScheme")
}
mode match {
case hdfsMode @ Hdfs(_, _) => readOrWrite match {
case Read => {
new HBaseRawTap(quorumNames, tableName, hBaseScheme, base64Scan, sinkMode match {
case null => SinkMode.KEEP
case _ => sinkMode
}).asInstanceOf[Tap[_, _, _]]
}
case Write => {
new HBaseRawTap(quorumNames, tableName, hBaseScheme, base64Scan, sinkMode match {
case null => SinkMode.UPDATE
case _ => sinkMode
}).asInstanceOf[Tap[_, _, _]]
}
}
case _ => super.createTap(readOrWrite)(mode)
}
}
}
|
