Ċ import ĠRetry Utility . retry Ċ import Ġio . prometheus . client . Counter Ċ import Ġio . prometheus . client . Summary Ċ import Ġorg . apache . hadoop . hbase .* Ċ import Ġorg . apache . hadoop . hbase . client . Connection Ċ import Ġorg . apache . hadoop . hbase . client . ConnectionFactory Ċ import Ġorg . apache . hadoop . hbase . client . Get Ċ import Ġorg . apache . hadoop . hbase . client . Put Ċ import Ġorg . apache . hadoop . hbase . io . Time Range Ċ import Ġorg . apache . hadoop . hbase . io . compress . Compression . Algorithm Ċ import Ġuk . gov . d wp . data works . logging . Data works Logger Ċ import Ġjava . io . IOException Ċ import Ġkotlin . system . measure Time Millis Ċ import Ġkotlin . time . Experimental Time Ċ Ċ @ Experimental Time Ċ open Ġclass ĠH base Client ( val Ġconnection : ĠConnection , Ġprivate Ġval Ġcolumn Family : ĠByteArray , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġcolumn Qualifier : ĠByteArray , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġh base Region Replication : ĠInt , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġsuc cesses : ĠSummary , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġretries : ĠCounter , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġfailures : ĠCounter ) Ġ: ĠAuto Closeable Ġ{ ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġopen Ġsuspend Ġfun Ġput List ( tableName : ĠString , Ġpayload s : ĠList < H base Payload >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( payload s . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtime Taken Ġ= Ġmeasure Time Millis Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Put ting Ġbatch Ġinto Ġtable ", Ġ" size " Ġto Ġ"${ payload s . size }", Ġ" table " Ġto ĠtableName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġensure Table ( tableName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretry ( suc cesses , Ġretries , Ġfailures , Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . getTable ( TableName . valueOf ( tableName )). use Ġ{ Ġtable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtable . put ( payload s . map Ġ{ Ġpayload Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPut ( payload . key ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Column ( column Family , Ġcolumn Qualifier , Ġpayload . version , Ġpayload . body ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, Ġpayload s [ 0 ]. record . topic (), Ġ"${ payload s [ 0 ]. record . partition () }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Put Ġbatch Ġinto Ġtable ", Ġ" time _ taken " Ġto Ġ"$ time Taken ", Ġ" size " Ġto Ġ"${ payload s . size }", Ġ" table " Ġto ĠtableName ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġput Record ( tableName : ĠString , Ġkey : ĠByteArray , Ġversion : ĠLong , Ġbody : ĠByteArray ) Ġ= ĊĠĠĠĠĠĠĠ Ġconnection . getTable ( TableName . valueOf ( tableName )). use Ġ{ Ġtable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtable . put ( Put ( key ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Column ( column Family , Ġcolumn Qualifier , Ġversion , Ġbody ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġget Cell After Timestamp ( tableName : ĠString , Ġkey : ĠByteArray , Ġtimestamp : ĠLong ): ĠByteArray ? Ġ{ ĊĠĠĠĠĠĠĠ Ġconnection . getTable ( TableName . valueOf ( tableName )). use Ġ{ Ġtable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġtable . get ( Get ( key ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Time Range ( timestamp , ĠTime Range . INITIAL _ MAX _ TIMESTAMP ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult . getValue ( column Family , Ġcolumn Qualifier ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget Cell Before Timestamp ( tableName : ĠString , Ġkey : ĠByteArray , Ġtimestamp : ĠLong ): ĠByteArray ? Ġ{ ĊĠĠĠĠĠĠĠ Ġconnection . getTable ( TableName . valueOf ( tableName )). use Ġ{ Ġtable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġtable . get ( Get ( key ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Time Range ( 0 , Ġtimestamp ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġresult . getValue ( column Family , Ġcolumn Qualifier ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Synchron ized ĊĠĠĠ Ġopen Ġfun Ġensure Table ( tableName : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdata TableName Ġ= ĠTable Name . valueOf ( tableName ) ĊĠĠĠĠĠĠĠ Ġval Ġnamespace Ġ= Ġdata TableName . namespace AsString ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! namespaces . contains ( namespace )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Creating Ġnamespace ", Ġ" namespace " Ġto Ġnamespace ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . admin . create Namespace ( Namespace Descriptor . create ( namespace ). build ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠNamespace Exist Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Namespace Ġalready Ġexists , Ġprobably Ġcreated Ġby Ġanother Ġprocess ", Ġ" namespace " Ġto Ġnamespace ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnamespaces [ namespace ] Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! tables . contains ( tableName )) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġh base Table Ġ= ĠH Table Descriptor ( data TableName ). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadd Family ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠH Column Descriptor ( column Family ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmax Versions Ġ= ĠInt . MAX _ VALUE ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmin Versions Ġ= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompression Type Ġ= ĠAlgorithm . GZ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomp action Compression Type Ġ= ĠAlgorithm . GZ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġregion Replication Ġ= Ġh base Region Replication ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Table In H base ( hbase Table ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Table In H base ( hbase Table : ĠH Table Descriptor ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġregion Spl its Ġ= ĠConfig . H base . region Spl its . toInt () ĊĠĠĠĠĠĠĠ Ġval Ġsplits Ġ= ĠRegion Key Splitter . calculate Spl its ( region Spl its ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Creating Ġtable ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" table _ name " Ġto Ġh base Table . name AsString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" region _ splits " Ġto Ġ"$ region Spl its ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġconnection . admin . createTable ( hbase Table , Ġsplits . to Typed Array ()) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠTable Exists Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Did n 't Ġcreate Ġtable , Ġtable Ġalready Ġexists , Ġprobably Ġcreated Ġby Ġanother Ġprocess ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" table _ name " Ġto Ġh base Table . name AsString ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtables [ hbase Table . name AsString ] Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġget Table Regions ( tableName : ĠTable Name ): ĠMutable List < H Region Info >? Ġ= Ġconnection . admin . getTable Regions ( tableName ) ĊĊĠĠĠ Ġprivate Ġval Ġnamespaces Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġext ant Namespaces Ġ= Ġmutable Map Of < String , ĠBoolean >() ĊĊĠĠĠĠĠĠĠ Ġconnection . admin . list Namespace Descriptors () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġext ant Namespaces [ it . name ] Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġext ant Namespaces ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġtables Ġby Ġlazy Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnames Ġ= Ġmutable Map Of < String , ĠBoolean >() ĊĊĠĠĠĠĠĠĠ Ġconnection . admin . list Table Names (). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnames [ it . name AsString ] Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġnames ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġconnect (): ĠH base Client Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" H base Ġconnection Ġconfiguration ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠH Constants . Z OO KE E PER _ Z NODE _ PARENT Ġto ĠConfig . H base . config . get ( H Constants . Z OO KE E PER _ Z NODE _ PARENT ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠH Constants . Z OO KE E PER _ QU OR UM Ġto ĠConfig . H base . config . get ( H Constants . Z OO KE E PER _ QU OR UM ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" hbase . zookeeper . port " Ġto ĠConfig . H base . config . get (" hbase . zookeeper . port ")) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠH base Client ( ConnectionFactory . create Connection ( H Base Configuration . create ( Config . H base . config )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . H base . column Family . toByteArray (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . H base . column Qualifier . toByteArray (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . H base . region Replication , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics Client . hbase Suc cesses , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics Client . hbase Retries , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics Client . hbase Failures ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠData works Logger . getLogger ( H base Client :: class ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ= Ġconnection . close () Ċ } Ċ