Apache HBase + Apache Hadoop + Xceivers

Introduktion

Vissa av konfigurationsegenskaperna som finns i Apache Hadoop har en direkt effekt på klienter, som Apache HBase. En av dessa egenskaper kallas "dfs.datanode.max.xcievers" och tillhör HDFS-delprojektet. Den definierar antalet serversidetrådar och – i viss mån – sockets som används för dataanslutningar. Att ställa in detta nummer för lågt kan orsaka problem när du växer eller öka användningen av ditt kluster. Det här inlägget hjälper dig att förstå vad som händer mellan klienten och servern och hur du bestämmer ett rimligt antal för den här egenskapen.

Problemet

Eftersom HBase lagrar allt den behöver inuti HDFS, kan den hårda övre gränsen som läggs upp av konfigurationsegenskapen "dfs.datanode.max.xcievers" resultera i att för få resurser är tillgängliga för HBase, vilket visar sig som IOExceptions på vardera sidan av anslutningen. Här är ett exempel från HBases e-postlista [1], där följande meddelanden först loggades på RegionServer-sidan:

2008-11-11 19:55:52,451 INFO org.apache.hadoop.dfs.DFSClient: Undantag i createBlockOutputStream java.io.IOException:Kunde inte läsa från ström
2008-11-11 19:55:52,451 INFO org.apache.hadoop.dfs.DFSClient: Abandoning block blk_-54670141087501_3907501_3907501_3907501_3957501 11 19:55:58,455 VARNA org.apache.hadoop.dfs.DFSClient: DataStreamer Undantag:java.io.IOException:Det går inte att skapa nytt block.
2008-11-11 19:55:58,455 VARNA org. .hadoop.dfs.DFSClient:Error Recovery for block blk_-5467014108758633036_595771 bad datanode[0]
2008-11-11 19:55:58,482 FATAL doop.apache.Region required h . Tvingar serveravstängning

Att korrelera detta med Hadoop DataNode-loggarna visade följande inlägg:

FEL org.apache.hadoop.dfs.DataNode: DatanodeRegistration(10.10.10.53:50010,storageID=DS-1570581820-10.10.10.53-50010-12241117842339,infoPort=50075, ipcPort=50020):DataXceiver:java.io.IOException of2

I det här exemplet orsakade det låga värdet på "dfs.datanode.max.xcievers" för DataNodes att hela RegionServer stängdes av. Det här är en riktigt dålig situation. Tyvärr finns det ingen hård-och-snabb regel som förklarar hur man beräknar den nödvändiga gränsen. Det rekommenderas vanligtvis att höja siffran från standardvärdet 256 till något i stil med 4096 (se [1], [2], [3], [4] och [5] för referens). Detta görs genom att lägga till den här egenskapen i filen hdfs-site.xml för alla DataNodes (observera att den är felstavad):

    dfs.datanode.max.xcievers
4096

Obs! Du måste starta om dina DataNodes efter att ha gjort denna ändring i konfigurationsfilen.

Det här borde hjälpa till med ovanstående problem, men du kanske fortfarande vill veta mer om hur allt detta spelar ihop och vad HBase gör med dessa resurser. Vi kommer att diskutera detta i resten av detta inlägg. Men innan vi gör det måste vi vara tydliga med varför du inte helt enkelt kan ställa in den här siffran väldigt högt, säg 64K och vara klar med det.

Det finns en anledning till en övre gräns, och den är tvåfaldig:för det första behöver trådar sin egen stack, vilket betyder att de upptar minnet. För nuvarande servrar innebär detta 1 MB per tråd[6] som standard. Med andra ord, om du använder alla 4096 DataXceiver-trådar behöver du cirka 4 GB hög för att rymma dem. Detta skär in i utrymmet du har tilldelat för memstores och blockcacher, såväl som alla andra rörliga delar av JVM. I värsta fall kan du stöta på en OutOfMemoryException, och RegionServer-processen är toast. Du vill ställa in den här egenskapen till ett rimligt högt värde, men inte för högt heller.

För det andra, med dessa många trådar aktiva kommer du också att se din CPU bli allt mer laddad. Det kommer att ske många sammanhangsbyten för att hantera allt samtidigt arbete, vilket tar bort resurser för det verkliga arbetet. Precis som med oron för minnet vill man att antalet trådar inte växer gränslöst, utan ger en rimlig övre gräns – och det är vad "dfs.datanode.max.xcievers" är till för.

Hadoop-filsystemdetaljer

Från klientsidan tillhandahåller HDFS-biblioteket abstraktionen som kallas Path. Den här klassen representerar en fil i ett filsystem som stöds av Hadoop, representerat av klassen FileSystem. Det finns några konkreta implementeringar av den abstrakta FileSystem-klassen, varav en är DistributedFileSytem, ​​som representerar HDFS. Denna klass omsluter i sin tur den faktiska DFSClient-klassen som hanterar all interaktion med fjärrservrarna, d.v.s. NameNode och de många DataNodes.

När en klient, som HBase, öppnar en fil gör den det genom att till exempel anropa metoderna open() eller create() i klassen FileSystem, här de mest förenklade inkarnationerna

  public DFSInputStream open(String src) throws IOException
public FSDataOutputStream create(Path f) throws IOException

Den returnerade stream-instansen är det som behöver en socket och tråd på serversidan, som används för att läsa och skriva datablock. De ingår i kontraktet för utbyte av data mellan klienten och servern. Observera att det finns andra, RPC-baserade protokoll som används mellan de olika maskinerna, men för denna diskussions syfte kan de ignoreras.

Ströminstansen som returneras är en specialiserad klass DFSOutputStream eller DFSInputStream, som hanterar all interaktion med NameNode för att ta reda på var kopiorna av blocken finns och datakommunikationen per block per DataNode.

På serversidan omsluter DataNode en instans av DataXceiverServer, som är den faktiska klassen som läser ovanstående konfigurationsnyckel och även kastar ovanstående undantag när gränsen överskrids.

När DataNode startar skapar den en trådgrupp och startar den nämnda DataXceiverServer instansen så här:

  this.threadGroup =new ThreadGroup(“dataXceiverServer”);
this.dataXceiverServer =new Daemon( threadGroup,
ny DataXceiverServer(ss, conf, this));
this.threadGroup.setDaemon(true); // förstör automatiskt när det är tomt

Observera att DataXceiverServer tråden redan tar upp en plats i trådgruppen. DataNode har också denna interna klass för att hämta antalet för närvarande aktiva trådar i denna grupp:

  /** Antal samtidiga xceivers per nod. */
int getXceiverCount() {
return threadGroup ==null ? 0 :threadGroup.activeCount();
}

Läs- och skrivblock, som initierats av klienten, gör att en anslutning skapas, som lindas in av DataXceiverServer-tråden i en DataXceiver instans. Under denna hand off skapas en tråd som registreras i ovanstående trådgrupp. Så för varje aktiv läs- och skrivoperation spåras en ny tråd på serversidan. Om antalet trådar i gruppen överskrider det konfigurerade maximivärdet kastas det nämnda undantaget och registreras i DataNodens loggar:

  if (curXceiverCount> dataXceiverServer.maxXceiverCount) {
kasta ny IOException(“xceiverCount ” + curXceiverCount
+ ” överskrider gränsen för samtidiga xcievers ”
+ dataXceiverServer.maxXceiverCount);
}

Konsekvenser för kunder

Nu är frågan, hur förhåller sig klientens läsning och skrivning till trådarna på serversidan. Innan vi går in på detaljerna dock, låt oss använda felsökningsinformationen som DataXceiver-klassen loggar när den skapas och stängs

  LOG.debug(“Antalet aktiva anslutningar är:” + datanode.getXceiverCount());
…
LOG.debug(datanode.dnRegistration + “:Antalet aktiva anslutningar är:”     + datanode.getXceiverCount());

och övervaka under en start av HBase vad som loggas på DataNode. För enkelhetens skull görs detta på en pseudodistribuerad installation med en enda DataNode och RegionServer-instans. Följande visar toppen av RegionServers statussida.

Den viktiga delen finns i avsnittet "Mätverk", där det står "storefiles=22". Så, om vi antar att HBase har åtminstone så många filer att hantera, plus några extra filer för loggen för att skriva framåt, bör vi se loggmeddelandet ovan ange att vi har minst 22 "aktiva anslutningar". Låt oss starta HBase och kontrollera loggfilerna för DataNode och RegionServer:

Kommandorad:

$ bin/start-hbase.sh
…

Datanodlogg:

2012-03-05 13:01:35,309 DEBUG org.apache.hadoop.hdfs.server.datanode. DataNode:Antalet aktiva anslutningar är:1
2012-03-05 13:01:35,315 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS- 1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:2
12/03/05 13:01:35 INFO3642449 globalMemStoreLimitLowMark=347.1m, maxHeap=991.7m
12/03/05 13:01:39 INFO http.HttpServer:Port som returneras av webServer.getConnectors()[0].getLocalPort() innan open() är -1 . Öppna lyssnaren på 60030
2012-03-05 13:01:40,003 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:1
12/03/05 13:01:40 INFO regionserver.HRegionServer:Mottagen begäran om att öppna region:-ROOT-,,0.70236052
2012-03-05 13:01:40,882 DEBUG org.apache.hadoop.hdfs.Dataserver.Nodenode. :Antalet aktiva anslutningar är:3
2012-03-05 13:01:40,884 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=6DS-44238 -10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:4
2012-03-05 13:01:40,888 DEBUG org.hdfsserver.ap datanode.DataNode:Antalet aktiva anslutningar är:3
…
12/03/05 13:01:40 INFO regionserver.HRegion:Onlined -ROOT-,,0.70236052; nästa sequenceid=63083
2012-03-05 13:01:40,982 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:3
2012-03-05 13; Antalet aktiva anslutningar är:4
…
12/03/05 13:01:41 INFO regionserver.HRegionServer:Mottagen begäran om att öppna region:.META.,,1.1028785192
2012-03 -05 13:01:41,026 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:3
2012-03-05 13:01:41,027 DEBUG org.apache.hadoop. hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075=50 av ipc:12/03/05 13:01:41 INFO regionserver.HRegion:Onlined .META.,,1.1028785192; nästa sequenceid=63082
2012-03-05 13:01:41,109 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:3
2012-03-05 13 :01:41,114 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:4
2012-03-05 13:01:41,117 DEBUG org.apache.hadoop.hdfs.server .datanode.DataNode:Antalet aktiva anslutningar är:5
12/03/05 13:01:41 INFO regionserver.HRegionServer:Mottagen begäran om att öppna 16 region(er)
12/03/05 13 :01:41 INFO regionserver.HRegionServer:Mottagen begäran om öppen region:usertable,,1330944810191.62a312d67981c86c42b6bc02e6ec7e3f.
12/03/05 13:030944810191.62a312d67981c86c42b6bc02e6ec7e3f.
12/03/05 13:01S 1330944810191.90d287473fe223f0ddc137020efda25d.
…

2012-03-05 13:01:41,246 DEBUG org.apache.hadoop.hdfs.server.datanode. DataNode:Antalet aktiva anslutningar är:6
2012-03-05 13:01:41,248 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:7
…
2012-03-05 13:01:41,257 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID=DS-14236420508-16236420500-1323-162-162-162-13236424500-13236420500-13236420500-13236420500-13236420500. , infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:10
2012-03-05 13:01:41,257 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0). 0.1:50010, lagrings-ID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:9
…/03/01 01:41 INFO regionserver.HRegion:Onlined usertable,user1120311784,1330944810191.90d287473fe223f0ddc137020efda25d.; nästa sekvensid=62917
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined usertable,,1330944810191.62a312d67981c86c42b6bc02e6ec7e3f.; nästa sekvensid=62916
…
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined usertable,user1361265841,1330944811370.80663fcf291e390969640f; nästa sequenceid=62919
2012-03-05 13:01:41,474 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:6
2012-03-05 13 :01:41,491 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:7
2012-03-05 13:01:41,495 DEBUG org.apache.hadoop.hdfs.server .datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=1423642020 aktiv anslutning:Number20s:3 -05 13:01:41,508 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:7
…
12/03/05 13:01:41 INFO regionserver .HRegion:Onlined usertable,user1964968041,1330944848231.dd89596e9129e1caa7e07f8a491c9734.; nästa sequenceid=62920
2012-03-05 13:01:41,618 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:6
2012-03-05 13 :01:41,621 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID=DS-1423642448-10.0.0.64-50010-23010-2321,070, 50010-23010, 2321, 50010-23210, 50010-23210, 50010-23010, 50010-23210, 0.64-50010-23010, 23010, 50010-23210, 0.64-50010-23010, 23010-23210, 50010-23210, 50010-2321 Antalet aktiva anslutningar är:7
…
2012-03-05 13:01:41,829 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID =DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:7
12/03/05 13:01:41 Serverbar online:region.HRt användare online:,user515290649,1330944849739.d23924dc9e9d5891f332c337977af83d.; nästa sequenceid=62926
2012-03-05 13:01:41,832 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:6
2012-03-05 13 :01:41,838 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, lagrings-ID=DS-1423642448-10.0.0.64-50010-2321,07010-23210,0.0.64-50010-23210, 0.0.64-50010-23210, 2321,0,0,0.64-50010-25210, 2521, 0, 50010, 2521, 50010, 2521, 50010, 2321, 50010, 2321, 50010, 2321, Antalet aktiva anslutningar är:7
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined usertable,user757669512,1330944850808.cd0d6f16d8ae9cf0c9277c6b9f.; nästa sequenceid=62929
…
2012-03-05 14:01:39,711 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:Antalet aktiva anslutningar är:4
2012 -03-05 22:48:41,945 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS-1423642448-10.0.0.135-1307-1270,327-13270,3270,3270,3270,3270,3270,3270,3270 ipcPort=50020):Antalet aktiva anslutningar är:4
12/03/05 22:48:41 INFO regionserver.HRegion:Onlined usertable,user757669512,1330944850808.cd0d6f16d0f8c9ae;cd0d6f16d0c9ae.cd5cd8c6967cd5cd6c96cdcf6c96cf6cf6c96cf6cdcf6c96cf6cf6cdf6c96cd nästa sequenceid=62929
2012-03-05 22:48:41,963 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS-144236.40.40236.4 -50010-1321352233772, infoPort=50075, ipcPort=50020):Antalet aktiva anslutningar är:4

Du kan se hur regionerna öppnas efter varandra, men vad du också kan märka är att antalet aktiva anslutningar aldrig stiger till 22 – det når knappt ens 10. Varför är det så? För att förstå detta bättre måste vi se hur filer i HDFS mappas till serversidans DataXceivers instans – och de faktiska trådarna de representerar.

Hadoop Deep Dive

De tidigare nämnda DFSInputStream och DFSOutputStream är verkligen fasader kring de vanliga streamkoncepten. De lindar in klient-serverkommunikationen i dessa standard Java-gränssnitt, samtidigt som de internt dirigerar trafiken till en vald DataNode – som är den som har en kopia av det aktuella blocket. Den har friheten att öppna och stänga dessa anslutningar efter behov. När en klient läser en fil i HDFS växlar klientbiblioteksklasserna transparent från block till block, och därför från DataNode till DataNode, så den måste öppna och stänga anslutningar efter behov.

 DFSInputStream har en instans av en DFSClient.BlockReader klass, som öppnar anslutningen till DataNode. Ströminstansen anropar blockSeekTo() för varje call to read() som tar hand om att öppna anslutningen, om det inte redan finns någon. När ett block är helt läst stängs anslutningen. Att stänga strömmen har naturligtvis samma effekt.

 DFSOutputStream har en liknande hjälparklass, DataStreamer. Den spårar anslutningen till servern, som initieras av nextBlockOutputStream()-metoden. Den har ytterligare interna klasser som hjälper till att skriva ut blockdata, som vi utelämnar här för korthetens skull.

Både skriv- och läsblock kräver en tråd för att hålla socket och mellanliggande data på serversidan, insvept i DataXceiver-instansen. Beroende på vad din klient gör, kommer du att se antalet anslutningar fluktuera runt antalet för närvarande åtkomliga filer i HDFS.

Tillbaka till HBase-gåtan ovan:anledningen till att du inte ser upp till 22 (och fler) anslutningar under starten är att medan regionerna är öppna, är den enda nödvändiga informationen HFiles infoblock. Detta block läses för att få viktiga detaljer om varje fil, men stängs sedan igen. Detta innebär att resursen på serversidan släpps i snabb följd. De återstående fyra kopplingarna är svårare att fastställa. Du kan använda JStack för att dumpa alla trådar på DataNode, som i detta exempel visar denna post:

“DataXceiver för klient /127.0.0.1:64281 [sändningsblock blk_5532741233443227208_4201]” daemon prio=504 tid=7 demon prio=506 tid=5 nid=0x1178b4000 körbar [1178b3000]
java.lang.Thread.State:RUNNABLE
…

“DataXceiver for client /127.0.0.1:64172 [receiving block blk_-2005512129579433420_4199_1 client=DFSCl.0sient=0.2SCl. ,60020,1330984111693_1330984118810]” daemon prio=5 tid=7fb966109000 nid=0x1169cb000 körbar [1169ca000]
java.lange
Tråden.ABLE

Detta är de enda DataXceiver-posterna (i det här exemplet), så antalet i trådgruppen är lite missvisande. Kom ihåg att DataXceiverServer-demontråden redan står för en extra post, som i kombination med de två ovanstående står för de tre aktiva anslutningarna – vilket i själva verket betyder tre aktiva trådar. Anledningen till att loggen anger fyra istället är att den loggar räkningen från en aktiv tråd som håller på att avslutas. Så kort efter att antalet fyra har loggats är det faktiskt en mindre, dvs tre och matchar därmed vårt antal aktiva trådar.

Notera också att de interna hjälpklasserna, såsom PacketResponder, upptar en annan tråd i gruppen medan de är aktiva. JStack-utgången indikerar detta faktum, listar tråden som sådan:

 “PacketResponder 0 for Block blk_-2005512129579433420_4199” daemon prio=5 tid=7fb96384d001 nidaceit=0x00 in Object. () [116acd000]
java.lang.Thread.State:TIMED_WAITING (på objektövervakare)
på java.lang.Object.wait(Native Method)
på org.apache.hadoop. hdfs.server.datanode.BlockReceiver$PacketResponder \
.lastDataNodeRun(BlockReceiver.java:779)
– låst (en org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder)
på org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder.run(BlockReceiver.java:870)
på java.lang.Thread.run(Thread.java:680)

Den här tråden är för närvarande i TIMED_WAITING tillstånd och anses inte vara aktiv. Det är därför räkningen som sänds ut av DataXceiver-loggsatserna inte inkluderar den här typen av trådar. Om de blir aktiva på grund av att klienten skickar sändningsdata kommer antalet aktiva trådar att gå upp igen. En annan sak att notera är att den här tråden inte behöver en separat anslutning, eller socket, mellan klienten och servern. PacketResponder är bara en tråd på serversidan för att ta emot blockdata och streama den till nästa DataNode i skrivpipelinen.

Hadoop fsck-kommandot har också ett alternativ för att rapportera vilka filer som för närvarande är öppna för skrivning:

$ hadoop fsck /hbase -openforwrite
FSCK startat av larsgeorge från /10.0.0.29 för sökväg / hbase på Mån Mar 05 22:59:47 CET 2012
……/hbase/.logs/10.0.0.29,60020,1330984111693/10.0.0.29%3A60020.13380948411 block, OPENITEs, OPEN ………………………………..Status:HEALTHY
Total storlek:     2088783626 B
Totalt dirs:     54
Totalt antal filer:   45
…

Detta hänför sig inte omedelbart till en upptagen tråd på serversidan, eftersom dessa allokeras av block-ID. Men du kan dra ur det att det finns ett öppet block för att skriva. Hadoop-kommandot har ytterligare alternativ för att skriva ut de faktiska filerna och block-ID:t de består av:

$ hadoop fsck /hbase -files -blocks
FSCK startat av larsgeorge från /10.0.0.29 för sökväg /hbase på Tis Mar 06 10:39:50 CET 2012

...
/hbase/.META./1028785192/.tmp