HBase Clusters Gegevenssynchronisatie met HashTable/SyncTable-tool

Replicatie (besproken in dit vorige blogartikel) is al een tijdje uitgebracht en is een van de meest gebruikte functies van Apache HBase. Het hebben van clusters die gegevens repliceren met verschillende peers is een veel voorkomende implementatie, of het nu gaat om een DR-strategie of gewoon als een naadloze manier om gegevens te repliceren tussen productie-/staging-/ontwikkelomgevingen. Hoewel het een efficiënte manier is om verschillende HBase-databases gesynchroniseerd te houden binnen een latentie van minder dan een seconde, werkt replicatie alleen over gegevens die zijn opgenomen nadat de functie is ingeschakeld. Dat betekent dat alle reeds bestaande gegevens over alle clusters die betrokken zijn bij de replicatie-implementatie nog steeds op een andere manier tussen de peers moeten worden gekopieerd. Er zijn nogal wat tools die kunnen worden gebruikt om reeds bestaande gegevens op verschillende peerclusters te synchroniseren. Snapshots, BulkLoad, CopyTable zijn bekende voorbeelden van dergelijke tools die in eerdere Cloudera-blogposts zijn behandeld. In dit artikel behandelen we HashTable/SyncTable, waarin enkele van de interne implementatielogica wordt beschreven, de voor- en nadelen van het gebruik ervan en hoe het zich verhoudt tot enkele van de andere hierboven genoemde technieken voor het kopiëren van gegevens.

HashTable/SyncTable in een notendop

HashTable/SyncTable is een tool die is geïmplementeerd als twee taken die de kaart verkleinen en die als afzonderlijke stappen worden uitgevoerd. Het lijkt op de CopyTable tool, die zowel gedeeltelijke als volledige tabelgegevenskopieën kan uitvoeren. In tegenstelling tot CopyTable het kopieert alleen afwijkende gegevens tussen doelclusters, waardoor zowel netwerk- als computerbronnen worden bespaard tijdens de kopieerprocedure.

De eerste stap die in het proces moet worden uitgevoerd, is de HashTable kaartverkleinen baan. Dit moet worden uitgevoerd op het cluster waarvan de gegevens moeten worden gekopieerd naar de externe peer, normaal gesproken het broncluster. Een snel voorbeeld van hoe u het moet uitvoeren, wordt hieronder weergegeven. Een gedetailleerde uitleg van elk van de vereiste parameters vindt u verderop in dit artikel:

hbase org.apache.hadoop.hbase.mapreduce.HashTable --families=cf my-table /hashes/test-tbl…20/04/28 05:05:48 INFO mapreduce.Job:  map 100% reduce 100 %20/04/28 05:05:49 INFO mapreduce.Job:Job job_1587986840019_0001 succesvol afgerond20/04/28 05:05:49 INFO mapreduce.Job:Tellers:68…File Input Format Tellers Bytes Read=0File Output Format Tellers Bytes Geschreven=6811788

Zodra de HashTable taakuitvoering met het bovenstaande commando is voltooid, enkele uitvoerbestanden zijn gegenereerd in de bron hdfs /hashes/my-table map:

hdfs dfs -ls -R /hashes/test-tbldrwxr-xr-x   - root supergroep          0 2020-04-28 05:05 /hashes/test-tbl/hashes-rw-r--r--   2 root supergroep          0 2020-04-28 05:05 /hashes/test-tbl/hashes/_SUCCESSdrwxr-xr-x   - root-supergroep          0 2020-04-28 05:05 /hashes/test-tbl/hashes/part-r-00000 -rw-r--r--   2 root-supergroep    6790909 2020-04-28 05:05 /hashes/test-tbl/hashes/part-r-00000/data-rw-r--r--   2 root-supergroep      20879 2020-04-28 05:05 /hashes/test-tbl/hashes/part-r-00000/index-rw-r--r--   2 root-supergroep         99 2020-04-28 05:04 /hashes/test- tbl/manifest-rw-r--r--   2 root-supergroep        153 2020-04-28 05:04 /hashes/test-tbl/partitions

Deze zijn nodig als invoer voor de SyncTable loop. SyncTable moet worden gelanceerd op de doelpeer. Onderstaande opdracht voert SyncTable uit voor de uitvoer van HashTable uit het vorige voorbeeld. Het maakt gebruik van de dryrun optie verderop in dit artikel uitgelegd:

hbase org.apache.hadoop.hbase.mapreduce.SyncTable --dryrun --sourcezkcluster=zk1.example.com,zk2.example.com,zk3.example.com:2181:/hbase hdfs://source- cluster-active-nn/hashes/test-tbl test-tbl test-tbl...org.apache.hadoop.hbase.mapreduce.SyncTable$SyncMapper$CounterBATCHES=97148HASHES_MATCHED=97146HASHES_NOT_MATCHED=2MATCHINGCELLS=1=17MATCHINGSEMISSROWS=17MATCHINGSEMROWS=2SRANGESELLWSOTCHES=2SMATCHINGSELLWSIT 1

Als snelle referentie kunt u de gegeven parameters op beide voorbeelden gewoon vervangen door uw werkelijke omgevingswaarden. De rest van dit artikel gaat dieper in op de implementatiedetails.

Waarom twee verschillende stappen?

Het belangrijkste doel van deze tool is om alleen gegevens te identificeren en te kopiëren die tussen de twee clusters ontbreken. HashTable werkt als een sharding-/indexeringstaak, analyseert batches van de tabelgegevens en genereert hash-indexen voor elk van deze batches. Dit zijn de uitvoer die is geschreven in de bestanden onder de /hashes/my-table hdfs-map doorgegeven als een van de taakparameters. Zoals eerder vermeld, is deze uitvoer vereist door de SyncTable functie. SyncTable scant de doeltabel lokaal in dezelfde batchgroottes als gebruikt door HashTable, en berekent ook hash-waarden voor deze batches met dezelfde functie die wordt gebruikt door HashTable. Vervolgens vergelijkt het de lokale batch-hash waarde met die uit de HashTable uitvoer. Als de hash-waarden gelijk zijn, betekent dit dat de hele batch identiek is in de twee clusters en dat er niets naar dat segment hoeft te worden gekopieerd. Anders opent het een scan voor de batch in het broncluster, waarbij wordt gecontroleerd of elk van de cellen al in het doelcluster bestaat, waarbij alleen de cellen worden gekopieerd die afwijken. Bij schaarse, enigszins verschillende datasets zou dit ertoe leiden dat er veel minder gegevens tussen de twee clusters worden gekopieerd. Er hoeven ook maar een klein aantal cellen in de bron te worden gescand om te controleren op mismatches.

Vereiste parameters

HashTable vereist slechts twee parameters:de tabelnaam en het uitvoerpad waar gerelateerde hashes en andere meta-infobestanden worden geschreven. SyncTable gebruikt HashTable output dir als input, samen met de tabelnamen in respectievelijk de bron en in het doelcluster. Aangezien we HashTable gebruiken /SyncTable om gegevens tussen externe clusters te verplaatsen, sourcezkcluster optie moet worden gedefinieerd voor SyncTable . Dit moet het quorumadres van de dierenverzorger van het broncluster zijn. In dit artikelvoorbeeld hadden we ook rechtstreeks verwezen naar het actieve namenode-adres van het broncluster, zodat SyncTable leest het hash-uitvoerbestand rechtstreeks uit het broncluster. U kunt ook HashTable uitvoer kan handmatig worden gekopieerd van het broncluster naar het externe cluster (met bijvoorbeeld distcp).

OPMERKING:werken met externe clusters onder verschillende kerberos-realms wordt alleen ondersteund vanaf CDH 6.2.1 en later.

Geavanceerde opties

Beide HashTable en SyncTable bieden extra optionele opties die kunnen worden afgestemd voor optimale resultaten.

HashTabel maakt het mogelijk om de gegevens te filteren op zowel rijsleutel als wijzigingstijd, met startrow/starttime, stoprow/stoptime eigenschappen resp. Het bereik van de dataset kan ook worden beperkt door versies en gezinnen eigenschappen. De batchgrootte eigenschap definieert de grootte van elk gedeelte dat wordt gehasht. Dit heeft direct invloed op de synchronisatieprestaties. In het geval van zeer weinig mismatches, kan het instellen van grotere batchgroottewaarden leiden tot betere prestaties, aangezien grotere delen van de dataset zouden worden genegeerd zonder dat ze door SyncTable hoeven te worden gescand.

SyncTable biedt een dryrun optie waarmee u een voorbeeld kunt bekijken van de wijzigingen die in het doel moeten worden toegepast.

SyncTable standaardgedrag is om de brongegevens aan de doelzijde te spiegelen, dus elke extra cel die aanwezig is in het doel maar afwezig is in de bron, wordt uiteindelijk verwijderd aan de doelzijde. Dat kan ongewenst zijn bij het synchroniseren van clusters onder een Active-Active-replicatieconfiguratie en in dergelijke gevallen doDeletes opties kunnen worden omgezet in onwaar, waarbij replicatie van verwijderingen op het doel wordt overgeslagen. Er is ook een vergelijkbare doPuts vlag voor gevallen waarin geen extra cellen in het doelcluster mogen worden ingevoegd.

De output analyseren

HashTabel voert een paar bestanden uit met meta-informatie voor SyncTable, die zijn echter niet door mensen leesbaar. Het voert geen wijzigingen uit op bestaande gegevens, dus gerelateerde informatie is van weinig belang voor een gebruikerscontext.

SyncTable is de stap die de wijzigingen echt op het doel toepast en het kan belangrijk zijn om de samenvatting ervan te bekijken voordat de gegevens van het doelcluster daadwerkelijk worden gewijzigd (zie dryrun hierboven genoemde optie). Het publiceert enkele relevante tellers aan het einde van de kaart om de uitvoering te verminderen. Als we naar de waarden uit het bovenstaande voorbeeld kijken, kunnen we zien dat er 97148 . waren partities gehasht (gerapporteerd door BATCHES counter), die SyncTable ontdekte afwijkingen in slechts twee ervan (volgens de HASHES_MATCHED en HASHES_NOT_MACTHED tellers). Bovendien, binnen de twee partities met verschillende hashes, 17 cellen meer dan 2 rijen kwamen overeen (zoals gerapporteerd door MATCHING_CELLS en MATCHING_ROWS, respectievelijk), maar er waren ook 2 rijen divergerend, op deze twee partities (volgens RANGESNOTMATCHED en ROWSWITHDIFFS ). Tot slot, SOURCEMISSINGCELLS en TARGETMISSINGCELLS vertel ons in detail of cellen alleen aanwezig waren op de bron- of doelcluster. In dit voorbeeld had het broncluster één cel die niet op het doel stond, maar het doel had ook een cel die niet op de bron stond. Sinds SyncTable werd uitgevoerd zonder dryrun op te geven optie en instelling doDeletes optie om false , de job heeft de extra cel in het doelcluster verwijderd en de extra cel gevonden in bron aan het doelcluster toegevoegd. Ervan uitgaande dat geen schrijfacties gebeuren op beide clusters, een daaropvolgende uitvoering van dezelfde SyncTable commando in het doelcluster zou geen verschillen vertonen:

hbase org.apache.hadoop.hbase.mapreduce.SyncTable --sourcezkcluster=zk1.example.com,zk2.example.com,zk3.example.com:2181:/hbase hdfs://nn:9000/hashes /test-tbl test-tbl test-tbl…org.apache.hadoop.hbase.mapreduce.SyncTable$SyncMapper$CounterBATCHES=97148HASHES_MATCHED=97148…

Toepasselijke scenario's

Gegevenssynchronisatie

Op het eerste gezicht, HashTable/SyncTable lijkt misschien te overlappen met de CopyTable tool, maar er zijn nog steeds specifieke scenario's waarin beide tools meer geschikt zouden zijn. Als een eerste vergelijkingsvoorbeeld, met behulp van HashTable/SyncTable voor het aanvankelijk laden van een tabel met 100.004 rijen en een totale gegevensgrootte van 5,17 GB waren enkele minuten nodig, alleen voor SyncTable voltooien:

...20/04/29 03:48:00 INFO mapreduce.Job:Lopende taak:job_1587985272792_001120/04/29 03:48:09 INFO mapreduce.Job:Job job_1587985272792_0011 draaien in uber-modus:false20/04/ 29 03:48:09 INFO mapreduce.Job: map 0% reduce 0%20/04/29 03:54:08 INFO mapreduce.Job: map 100% reduce 0%20/04/29 03:54:09 INFO mapreduce .Job:Job job_1587985272792_0011 met succes voltooid...org.apache.hadoop.hbase.mapreduce.SyncTable$SyncMapper$CounterBATCHES=97148EMPTY_BATCHES=97148HASHES_NOT_MATCHED=97148RANGESNOTMATCHED=97148ROWS1000WITHDIFFSTM=97148

Zelfs op zo'n kleine dataset, CopyTable sneller uitgevoerd (ongeveer 3 minuten, terwijl SyncTable duurde 6 minuten om de hele dataset te kopiëren):

...20/04/29 05:12:07 INFO mapreduce.Job:Lopende taak:job_1587986840019_000520/04/29 05:12:24 INFO mapreduce.Job:Job job_1587986840019_0005 in uber-modus:false20/04/ 29 05:12:24 INFO mapreduce.Job:  map 0% reduce 0%20/04/29 05:13:16 INFO mapreduce.Job:  map 25% reduce 0%20/04/29 05:13:49 INFO mapreduce .Job:  map 50% verminderen 0%20/04/29 05:14:37 INFO mapreduce.Job:  map 75% verminderen 0%20/04/29 05:15:14 INFO mapreduce.Job:  map 100% verminderen 0 %20/04/29 05:15:14 INFO mapreduce.Job:Job job_1587986840019_0005 completed successfully…HBase CountersBYTES_IN_REMOTE_RESULTS=2787236791BYTES_IN_RESULTS=5549784428MILLIS_BETWEEN_NEXTS=130808NOT_SERVING_REGION_EXCEPTION=0NUM_SCANNER_RESTARTS=0NUM_SCAN_RESULTS_STALE=0REGIONS_SCANNED=4REMOTE_RPC_CALLS=1334REMOTE_RPC_RETRIES=0ROWS_FILTERED=0ROWS_SCANNED=100004RPC_CALLS=2657RPC_RETRIES=0

Laten we nu beide tools opnieuw gebruiken om kleine verschillen over de dataset op te lossen. De test-tbl tabel die in al deze voorbeelden wordt gebruikt, heeft vier regio's in het broncluster. Nadat in het vorige voorbeeld alle originele datasets naar het doelcluster waren gekopieerd, hebben we alleen aan de bronzijde vier extra rijen toegevoegd, één voor elk van de bestaande regio's, en vervolgens HashTable/SyncTable uitgevoerd. opnieuw om te synchroniseren beide clusters:

20/04/29 05:29:23 INFO mapreduce.Job:Lopende job:job_1587985272792_001320/04/29 05:29:39 INFO mapreduce.Job:Job job_1587985272792_0013 draaien in uber-modus:false20/04/29 05:29:39 INFO mapreduce.Job:  map 0% reduce 0%20/04/29 05:29:53 INFO mapreduce.Job:  map 50% reduce 0%20/04/29 05:30:42 INFO mapreduce.Job:kaart 100% verminderen 0%20/04/29 05:30:42 INFO mapreduce.Job:Job job_1587985272792_0013 succesvol afgerond...org.apache.hadoop.hbase.mapreduce.SyncTable$SyncMapper$CounterBATCHES=97148HASHES_MATCHED=97144HASHES_NOMATCH_MATCHED=4=MATCHMATCHINGROELLS 5RANGESNOTMATCHED=4ROWSWITHDIFFS=4TARGETMISSINGCELLS=4TARGETMISSINGROWS=4

We kunnen dat zien met slechts vier partities komen niet overeen, SyncTable was aanzienlijk sneller (ongeveer een minuut in beslag). CopyTable gebruiken om dezelfde synchronisatie uit te voeren toonde de volgende resultaten:

20/04/29 08:32:38 INFO mapreduce.Job:Lopende job:job_1587986840019_000820/04/29 08:32:52 INFO mapreduce.Job:Job job_1587986840019_0008 draaien in uber-modus:false20/04/29 08:32:52 INFO mapreduce.Job:  map 0% reduce 0%20/04/29 08:33:38 INFO mapreduce.Job:  map 25% reduce 0%20/04/29 08:34:15 INFO mapreduce.Job:kaart 50% verminderen 0%20/04/29 08:34:48 INFO mapreduce.Job:  map 75% verminderen 0%20/04/29 08:35:31 INFO mapreduce.Job:  map 100% verminderen 0%20/ 04/29 08:35:32 INFO mapreduce.Job:Job job_1587986840019_0008 completed successfully…HBase CountersBYTES_IN_REMOTE_RESULTS=2762547723BYTES_IN_RESULTS=5549784600MILLIS_BETWEEN_NEXTS=340672NOT_SERVING_REGION_EXCEPTION=0NUM_SCANNER_RESTARTS=0NUM_SCAN_RESULTS_STALE=0REGIONS_SCANNED=4REMOTE_RPC_CALLS=1323REMOTE_RPC_RETRIES=0ROWS_FILTERED=0ROWS_SCANNED=100008RPC_CALLS=2657RPC_RETRIES=0

Kopieertabel kostte evenveel tijd om de tabellen te synchroniseren als bij het kopiëren van de hele dataset, ook al waren er slechts vier cellen te kopiëren. Dit was nog steeds oké voor deze zeer kleine dataset, en met een inactief cluster, maar onder productiegebruik met grotere datasets en waar het doelcluster ook in gebruik kan zijn door veel clientapplicaties die er gegevens op schrijven, CopyTable prestatievermindering in vergelijking met SyncTable nog hoger zou zijn.

Het is vermeldenswaard dat er ook extra tools/functies zijn die in combinatie kunnen worden gebruikt voor een eerste lading van een doelcluster (het doel heeft helemaal geen gegevens), zoals het exporteren van snapshots, bulkladen of zelfs een directe kopie van het origineel tabel dirs van broncluster. Voor initiële ladingen met grote hoeveelheden gegevens om te kopiëren, maakt u een momentopname van een tabel en gebruikt u vervolgens de ExportSnapshot tool presteert beter dan online kopieertools zoals SyncTable of CopyTable.

Replicatie-integriteit controleren

Een ander veelgebruikt gebruik van HashTable/SyncTable is voor het bewaken van de replicatiestatus tussen clusters, bij het oplossen van mogelijke replicatieproblemen. In dit scenario functioneert het als een alternatief voor het hulpprogramma VerifyReplication. Meestal zijn er bij het controleren van de status tussen twee clusters of helemaal geen mismatches of heeft een tijdelijk optioneel probleem ervoor gezorgd dat een klein deel van de grotere dataset niet meer synchroon loopt. In de testomgeving die we voor ons vorige voorbeeld hebben gebruikt, moeten er 100.000 rijen zijn met overeenkomende waarden op beide clusters. Door SyncTable op het bestemmingscluster uit te voeren met de optie 'dryrun' kunnen we eventuele verschillen identificeren:

20/05/04 10:47:25 INFO mapreduce.Job:Lopende taak:job_1588611199158_0004…20/05/04 10:48:48 INFO mapreduce.Job:  map 100% verminderen 0%20/05/04 10 :48:48 INFO mapreduce.Job:Taak job_1588611199158_0004 succesvol voltooid...HBase CountersBYTES_IN_REMOTE_RESULTS=3753476784BYTES_IN_RESULTS=5549784600ROWS_SCANNED=100008...org.apache.hadoop.hbase.mapreduce... het hulpprogramma VerifyReplication op het broncluster. We geven de peer-ID door als een van de parameters, zodat deze de externe cluster kan vinden om te scannen voor vergelijking:20/05/04 11:01:58 INFO mapreduce.Job:Lopende taak:job_1588611196128_0001...20/05/04 11:04:39 INFO mapreduce.Job:kaart 100% verminderen 0%20/05/04 11:04:39 INFO mapreduce.Job:Job job_1588611196128_0001 succesvol voltooid...HBase CountersBYTES_IN_REMOTE_RESULTS=2761955495BYTES_IN_RESULTS=5549784600...org.aphache.maphadodoorg.apache .replication.VerifyReplication$Verifier$CountersGOODROWS=100008...

Zonder verschillen vindt SyncTable alle hashes die overeenkomen tussen bron- en doelpartities en vermijdt zo de noodzaak om het externe broncluster opnieuw te scannen. VerifyReplication voert een één voor één vergelijking uit voor elke cel in beide clusters, die mogelijk al hoge netwerkkosten met zich meebrengen, zelfs bij het omgaan met dergelijke kleine datasets.

Een extra rij toevoegen in het broncluster en de controles opnieuw uitvoeren. Met VerifyReplication:

20/05/05 11:14:05 INFO mapreduce.Job:Lopende job:job_1588611196128_0004…20/05/05 11:16:32 INFO mapreduce.Job:  map 100% verminderen 0%20/05/05 11 :16:32 INFO mapreduce.Job:Job job_1588611196128_0004 succesvol afgerond...org.apache.hadoop.hbase.mapreduce.replication.VerifyReplication$Verifier$CountersBADROWS=1GOODROWS=100008ONLY_IN_SOURCE_TABLE_ROWS=1…

Voordat we SyncTable kunnen gebruiken, moeten we hashes op de bron opnieuw genereren met HashTable, omdat er nu een nieuwe cel is:

20/05/04 11:31:48 INFO mapreduce.Job:Lopende job:job_1588611196128_0003…20/05/04 11:33:15 INFO mapreduce.Job:Job job_1588611196128_0003 succesvol afgerond...

Nu SyncTable:

20/05/07 05:47:51 INFO mapreduce.Job:Lopende job:job_1588611199158_0014…20/05/07 05:49:20 INFO mapreduce.Job:Job job_1588611199158_0014 succesvol afgerond… org.apache.hadoop.hbase.mapreduce.SyncTable$SyncMapper$CounterBATCHES=97148HASHES_NOT_MATCHED=97148MATCHINGCELLS=749593MATCHINGROWS=100008RANGESMATCHED=97147RANGESNOTMATCHED=1ROWSWITHDISSINGCELLRINGROGETM>

We kunnen de toename van de uitvoeringstijd zien als gevolg van extra scan- en celvergelijking tussen de twee afgelegen clusters. Ondertussen vertoonde de uitvoeringstijd van VerifyReplication weinig variatie.

Conclusie

HashTable/SyncTable is een waardevol hulpmiddel voor het verplaatsen van gegevens bij het omgaan met schaarse mismatches tussen twee clusters-gegevenssets. Het maakt gebruik van gegevenspartitionering en hashing om efficiënt verschillen in bereiken van de twee gegevenssets te detecteren, waardoor het aantal te scannen cellen wordt verminderd terwijl gegevens uit de twee clusters worden vergeleken, terwijl ook onnodige plaatsing van reeds bestaande waarden in het doelcluster wordt vermeden. Het is echter geen wondermiddel, zoals blijkt uit enkele van de bovenstaande voorbeelden, terwijl het lijkt te overlappen in functie met de CopyTable tool, kan de laatste betere prestaties bieden bij het hanteren van een groter, opeenvolgend bereik van niet-overeenkomende cellen. In die zin is HashTable/SyncTable zou het meest van toepassing zijn in gevallen waarin beide clusters al gegevens bevatten, of in gevallen waarin een bestaande replicatie-installatie is verstoord door tijdelijke onbeschikbaarheid van een van de peers.