· 8 years ago · Mar 28, 2018, 02:10 PM
1import cats.effect._
2import cats.implicits._
3import fs2._
4import org.apache.hadoop.conf.Configuration
5import org.apache.hadoop.hbase.client._
6import org.apache.hadoop.hbase.util.Bytes
7import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor}
8import org.log4s.getLogger
9import org.joda.time.DateTime
10
11object HBaseDB {
12
13 // This is the example function that takes the pool and constructs the stream of results.
14 // Likely you would like you units of Computation like the scan and table name passed as
15 // parameters and then you can operate on functions on each result from the returning
16 // stream.
17 // My apologies that this API is out of date. Some adjustments will likely need to
18 // be made with the modern hbase api.
19 def exampleFunction[F[_]](hc: HTablePool)(implicit F: Sync[F]): Stream[F, Result] = {
20 import scala.collection.JavaConverters._
21 for {
22 hti <- makeTableInterface(hc, ???) // Table Name
23 scan <- Stream.eval(F.delay(???)) // Construct Scan
24 result <- Stream.bracket(F.delay(hti.getScanner(scan)))(
25 {s => Stream.fromIterator(s.iterator().asScala).covary[F]},
26 scanner => F.delay(scanner.close())
27 )
28 } yield result
29 }
30
31
32
33 val logger = getLogger
34
35 def createConfiguration(quorum: String, master: String, clientPort: String, tcpNoDelay: String): Configuration = {
36 new Configuration(false) {
37 set("hbase.zookeeper.quorum", quorum)
38 set("hbase.master", master)
39 set("hbase.zookeeper.property.clientPort", clientPort)
40 set("hbase.ipc.client.tcpnodelay", tcpNoDelay)
41 }
42 }
43
44 def createPool[F[_]](configuration: Configuration)(implicit F: Sync[F]): Stream[F, HTablePool] =
45 Stream.bracket(
46 F.delay(logger.trace("Creating HTablePool")) *>
47 F.delay(new HTablePool(configuration, Int.MaxValue))
48 )(
49 Stream.emit(_).covary[F],
50 hc => F.delay(logger.trace("Closing HTablePool")) *> F.delay(hc.close())
51 )
52
53 def createAdmin[F[_]](configuration: Configuration)(implicit F: Sync[F]): Stream[F, HBaseAdmin] =
54 Stream.bracket(
55 F.delay(logger.trace("Creating HBaseAdmin")) *>
56 F.delay{
57 val admin = new HBaseAdmin(configuration)
58 logger.trace("Admin Created")
59 admin
60 }
61 )(
62 Stream.emit(_).covary[F],
63 a => F.delay(logger.trace("Closing HBaseAdmin")) *> F.delay(a.close())
64 )
65
66 def makeTableInterface[F[_]](htablePool: HTablePool, tableName: String)(implicit F: Sync[F]): Stream[F, HTableInterface] =
67 Stream.bracket(
68 F.delay(logger.trace(s"Creating HTableInterface for $tableName")) *>
69 F.delay(htablePool.getTable(tableName)).attempt
70 .flatMap(_.fold(
71 e => F.delay(logger.error(e)(s"Failed to Create Table Interface $tableName")) *> F.raiseError[HTableInterface](e),
72 hti => F.delay(logger.trace(s"Created Table Interface for $tableName")).as(hti)
73 ))
74 )(
75 Stream.emit(_).covary[F],
76 interface =>
77 F.delay(logger.trace(s"Closing HTable Interface for $tableName")) *>
78 F.delay(interface.close())
79 )
80
81 def insertPut[F[_]](hTableInterface: HTableInterface, put: Put)(implicit F: Sync[F]): F[Unit] = F.delay{
82 hTableInterface.put(put)
83 }
84
85 def insertPuts[F[_]](htableInterface: HTableInterface, puts: List[Put])(implicit F: Sync[F]): F[Unit] = F.delay {
86 import scala.collection.JavaConverters._
87 htableInterface.put(puts.asJava)
88 }
89
90 def createTableWithColumns[F[_]](admin : HBaseAdmin, tableName: String, columnFamiles: String*)(implicit F: Sync[F]): F[Unit] = {
91 for {
92 _ <- createTable(admin, tableName)
93 result <- createColumns(admin, tableName, columnFamiles:_*)
94 } yield result
95 }
96
97 def createColumns[F[_]](hBaseAdmin: HBaseAdmin, tableName: String, columnFamilies: String*)(implicit F: Sync[F]): F[Unit] =
98 for {
99 _ <- disableTable(hBaseAdmin, tableName)
100 _ <- columnFamilies.toList.traverse(descriptor => F.delay(hBaseAdmin.addColumn(tableName, new HColumnDescriptor(descriptor))))
101 result <- enableTable(hBaseAdmin, tableName)
102 } yield result
103
104 def createTable[F[_]](hBaseAdmin: HBaseAdmin, tableName: String)(implicit F: Sync[F]): F[Unit] = {
105 F.delay(hBaseAdmin.tableExists(tableName))
106 .ifM(
107 F.pure(()),
108 F.delay(logger.trace(s"Creating Table $tableName")) *>
109 F.delay(hBaseAdmin.createTable(new HTableDescriptor(Bytes.toBytes(tableName))))
110 )
111 }
112
113 def deleteTable[F[_]](hBaseAdmin: HBaseAdmin, tableName: String)(implicit F: Sync[F]): F[Unit] = {
114 F.delay(hBaseAdmin.tableExists(tableName))
115 .flatMap(bool => F.delay(logger.trace(s"Checking if Table Exists - $tableName - $bool")).as(bool))
116 .ifM(
117 F.delay(logger.trace(s"Deleting Table $tableName")) *>
118 disableTable[F](hBaseAdmin, tableName) *>
119 F.delay(hBaseAdmin.deleteTable(tableName)),
120 F.delay(logger.trace(s"Attempted to Delete Table - $tableName - Does Not Exist"))
121 )
122 }
123
124 def enableTable[F[_]](hBaseAdmin: HBaseAdmin, tableName: String)(implicit F: Sync[F]): F[Unit] = {
125 F.delay(hBaseAdmin.isTableDisabled(tableName))
126 .ifM(
127 F.delay(logger.trace(s"Enabling Table $tableName")) *> F.delay(hBaseAdmin.enableTable(tableName)),
128 F.pure(())
129 )
130 }
131
132 def disableTable[F[_]](hBaseAdmin: HBaseAdmin, tableName: String)(implicit F: Sync[F]): F[Unit] = {
133 F.delay(hBaseAdmin.isTableDisabled(tableName))
134 .ifM(
135 F.pure(()),
136 F.delay(logger.trace(s"Disabling Table $tableName")) *> F.delay(hBaseAdmin.disableTable(tableName))
137 )
138 }
139
140
141
142}