Apache HBase + Apache Hadoop + Xceivers

Introducción

Algunas de las propiedades de configuración que se encuentran en Apache Hadoop tienen un efecto directo en los clientes, como Apache HBase. Una de esas propiedades se llama "dfs.datanode.max.xcievers" y pertenece al subproyecto HDFS. Define la cantidad de subprocesos del lado del servidor y, hasta cierto punto, los sockets utilizados para las conexiones de datos. Establecer este número demasiado bajo puede causar problemas a medida que crece o aumenta la utilización de su clúster. Esta publicación lo ayudará a comprender qué sucede entre el cliente y el servidor, y cómo determinar un número razonable para esta propiedad.

El problema

Dado que HBase almacena todo lo que necesita dentro de HDFS, el límite superior estricto impuesto por la propiedad de configuración "dfs.datanode.max.xcievers" puede dar como resultado que haya muy pocos recursos disponibles para HBase, lo que se manifiesta como IOExceptions en cualquier lado de la conexión. Este es un ejemplo de la lista de correo de HBase [1], donde los siguientes mensajes se registraron inicialmente en el lado de RegionServer:

2008-11-11 19:55:52,451 INFO org.apache.hadoop.dfs.DFSClient:excepción en createBlockOutputStream java.io.IOException:no se pudo leer de la secuencia
2008-11-11 19:55:52,451 INFO org.apache.hadoop.dfs.DFSClient: bloqueo abandonado blk_-5467014108758633036_595771
2008-11- 11 19:55:58,455 WARN org.apache.hadoop.dfs.DFSClient: Excepción de DataStreamer:java.io.IOException:No se puede crear un nuevo bloque.
2008-11-11 19:55:58,455 WARN org.apache .hadoop.dfs.DFSClient:Recuperación de error para el bloque blk_-5467014108758633036_595771 bad datanode[0]
2008-11-11 19:55:58,482 FATAL org.apache.hadoop.hbase.regionserver.Flusher:Repetición de hlog requerida . Forzar el apagado del servidor

La correlación de esto con los registros de Hadoop DataNode reveló la siguiente entrada:

ERROR org.apache.hadoop.dfs.DataNode: DatanodeRegistration(10.10.10.53:50010,storageID=DS-1570581820-10.10.10.53-50010-1224117842339,infoPort=50075, ipcPort=50020):DataXceiver:java.io.IOException: xceiverCount 258 supera el límite de xcievers simultáneos 256

En este ejemplo, el bajo valor de “dfs.datanode.max.xcievers” para los DataNodes hizo que todo el RegionServer se cerrara. Esta es una situación realmente mala. Desafortunadamente, no existe una regla estricta que explique cómo calcular el límite requerido. Por lo general, se recomienda aumentar el número predeterminado de 256 a algo así como 4096 (consulte [1], [2], [3], [4] y [5] como referencia). Esto se hace agregando esta propiedad al archivo hdfs-site.xml de todos los DataNodes (tenga en cuenta que está mal escrito):

    dfs.datanode.max.xcievers
4096

Nota:deberá reiniciar sus DataNodes después de realizar este cambio en el archivo de configuración.

Esto debería ayudar con el problema anterior, pero es posible que desee saber más sobre cómo funciona todo esto en conjunto y qué está haciendo HBase con estos recursos. Discutiremos esto en el resto de esta publicación. Pero antes de hacerlo, debemos aclarar por qué no puede simplemente configurar este número muy alto, digamos 64K y listo.

Hay una razón para un límite superior, y es doble:primero, los subprocesos necesitan su propia pila, lo que significa que ocupan memoria. Para los servidores actuales, esto significa 1 MB por subproceso[6] de forma predeterminada. En otras palabras, si usa todos los hilos 4096 DataXceiver, necesita alrededor de 4 GB de almacenamiento dinámico para acomodarlos. Esto corta el espacio que ha asignado para memstores y cachés de bloque, así como todas las demás partes móviles de la JVM. En el peor de los casos, es posible que se encuentre con una excepción OutOfMemoryException y el proceso de RegionServer se brinde. Desea establecer esta propiedad en un número razonablemente alto, pero tampoco demasiado alto.

En segundo lugar, al tener estos muchos subprocesos activos, también verá que su CPU se carga cada vez más. Habrá muchos cambios de contexto para manejar todo el trabajo concurrente, lo que quita recursos para el trabajo real. Al igual que con las preocupaciones sobre la memoria, desea que la cantidad de subprocesos no crezca sin límites, sino que proporcione un límite superior razonable, y para eso está "dfs.datanode.max.xcievers".

Detalles del sistema de archivos Hadoop

Desde el lado del cliente, la biblioteca HDFS proporciona la abstracción llamada Ruta. Esta clase representa un archivo en un sistema de archivos compatible con Hadoop, representado por la clase FileSystem. Hay algunas implementaciones concretas de la clase abstracta FileSystem, una de las cuales es DistributedFileSytem, ​​que representa HDFS. Esta clase, a su vez, envuelve la clase DFSClient real que maneja todas las interacciones con los servidores remotos, es decir, el NameNode y los muchos DataNodes.

Cuando un cliente, como HBase, abre un archivo, lo hace, por ejemplo, llamando a los métodos open() o create() de la clase FileSystem, aquí las encarnaciones más simples

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

La instancia de transmisión devuelta es lo que necesita un socket y un subproceso del lado del servidor, que se utilizan para leer y escribir bloques de datos. Forman parte del contrato de intercambio de datos entre el cliente y el servidor. Tenga en cuenta que hay otros protocolos basados ​​en RPC en uso entre las distintas máquinas, pero para el propósito de esta discusión pueden ignorarse.

La instancia de flujo devuelta es una clase DFSOutputStream o DFSInputStream especializada, que maneja toda la interacción con NameNode para averiguar dónde residen las copias de los bloques y la comunicación de datos por bloque por DataNode.

En el lado del servidor, DataNode envuelve una instancia de DataXceiverServer, que es la clase real que lee la clave de configuración anterior y también lanza la excepción anterior cuando se excede el límite.

Cuando se inicia DataNode, crea un grupo de subprocesos e inicia la instancia de DataXceiverServer mencionada de la siguiente manera:

  this.threadGroup =new ThreadGroup(“dataXceiverServer”);
this.dataXceiverServer =new Daemon( threadGroup,
nuevo DataXceiverServer(ss, conf, this));
this.threadGroup.setDaemon(true); // destrucción automática cuando está vacío

Tenga en cuenta que el subproceso DataXceiverServer ya está ocupando un lugar del grupo de subprocesos. El DataNode también tiene esta clase interna para recuperar el número de subprocesos actualmente activos en este grupo:

  /** Número de xceivers simultáneos por nodo. */
int getXceiverCount() {
return threadGroup ==null ? 0 :threadGroup.activeCount();
}

Los bloques de lectura y escritura, iniciados por el cliente, hacen que se establezca una conexión, que el subproceso DataXceiverServer envuelve en una instancia DataXceiver . Durante este traspaso, se crea y registra un subproceso en el grupo de subprocesos anterior. Entonces, para cada operación activa de lectura y escritura, se rastrea un nuevo hilo en el lado del servidor. Si el recuento de subprocesos en el grupo excede el máximo configurado, dicha excepción se lanza y se registra en los registros de DataNode:

  if (curXceiverCount> dataXceiverServer.maxXceiverCount) {
lanzar una nueva IOException(“xceiverCount ” + curXceiverCount
+ ” excede el límite de xcievers concurrentes ”
+ dataXceiverServer.maxXceiverCount);
}

Implicaciones para los clientes

Ahora, la pregunta es, ¿cómo se relaciona la lectura y escritura del cliente con los subprocesos del lado del servidor? Sin embargo, antes de entrar en detalles, usemos la información de depuración que la clase DataXceiver registra cuando se crea y se cierra

  LOG.debug(“El número de conexiones activas es:” + datanode.getXceiverCount());
…
LOG.debug(datanode.dnRegistration + “:El número de conexiones activas es:”     + datanode.getXceiverCount());

y monitoree durante un inicio de HBase lo que se registra en el DataNode. En aras de la simplicidad, esto se hace en una configuración pseudo distribuida con una sola instancia de DataNode y RegionServer. A continuación se muestra la parte superior de la página de estado de RegionServer.

La parte importante está en la sección “Métricas”, donde dice “storefiles=22”. Entonces, suponiendo que HBase tiene al menos esa cantidad de archivos para manejar, además de algunos archivos adicionales para el registro de escritura anticipada, deberíamos ver el mensaje de registro anterior que indica que tenemos al menos 22 "conexiones activas". Iniciemos HBase y verifiquemos los archivos de registro de DataNode y RegionServer:

Línea de comando:

$ bin/start-hbase.sh
…

Registro de nodo de datos:

2012-03-05 13:01:35,309 DEBUG org.apache.hadoop.hdfs.server.datanode. DataNode:el número de conexiones activas es: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):El número de conexiones activas es:2
12/03/05 13:01:35 INFO regionserver.MemStoreFlusher:globalMemStoreLimit=396.7m, globalMemStoreLimitLowMark=347.1m, maxHeap=991.7m
12/03/05 13:01:39 INFO http.HttpServer:Puerto devuelto por webServer.getConnectors()[0].getLocalPort() antes de open() es -1 . Abriendo el oyente en 60030
2012-03-05 13:01:40,003 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:1
12/03/05 13:01:40 INFO regionserver.HRegionServer:solicitud recibida para abrir la región:-ROOT-,,0.70236052
2012-03-05 13:01:40,882 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode :El número de conexiones activas es:3
2012-03-05 13:01:40,884 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):El número de conexiones activas es:4
2012-03-05 13:01:40,888 DEBUG org.apache.hadoop.hdfs.server. datanode.DataNode:el número de conexiones activas es:3
…
12/03/05 13:01:40 INFO regionserver.HRegion:Onlined -ROOT-,,0.70236052; siguiente secuenciaid=63083
2012-03-05 13:01:40,982 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:3
2012-03-05 13 :01:40,983 DEPURACIÓN org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipc0Port=5002):El número de conexiones activas es:4
…
12/03/05 13:01:41 INFO Regionserver.HRegionServer:Solicitud recibida para abrir región:.META.,,1.1028785192
2012-03 -05 13:01:41,026 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:3
2012-03-05 13:01:41,027 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):Número de conexiones activas:4
…
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined .META.,,1.1028785192; siguiente secuenciaid=63082
2012-03-05 13:01:41,109 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:3
2012-03-05 13 :01:41,114 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:4
2012-03-05 13:01:41,117 DEBUG org.apache.hadoop.hdfs.server .datanode.DataNode:el número de conexiones activas es:5
12/03/05 13:01:41 INFO Regionserver.HRegionServer:solicitud recibida para abrir 16 regiones
12/03/05 13 :01:41 INFO regionserver.HRegionServer:Solicitud recibida para abrir región:usertable,,1330944810191.62a312d67981c86c42b6bc02e6ec7e3f.
12/03/05 13:01:41 INFO regionserver.HRegionServer:Recibida solicitud para abrir región:usertable,user1840,311 1330944810191.90d287473fe223f0ddc137020efda25d.
…

2012-03-05 13:01:41,246 DEBUG org.apache.hadoop.hdfs.server.datanode. DataNode:el número de conexiones activas es:6
2012-03-05 13:01:41,248 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:7
…
2012-03-05 13:01:41,257 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):El número de conexiones activas es:10
2012-03-05 13:01:41,257 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):El número de conexiones activas es:9
…
12/03/05 13:01:41 INFO regionserver.HRegion:Tabla de usuario en línea, usuario1120311784,1330944810191.90d287473fe223f0ddc137020efda25d.; siguiente secuenciaid=62917
12/03/05 13:01:41 INFO regionserver.HRegion:tabla de usuario en línea, 1330944810191.62a312d67981c86c42b6bc02e6ec7e3f.; siguiente secuenciaid=62916
…
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined usertable,user1361265841,1330944811370.80663fcf291e3ce00080599964f406ba.; siguiente secuenciaid=62919
2012-03-05 13:01:41,474 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:6
2012-03-05 13 :01:41,491 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:7
2012-03-05 13:01:41,495 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):El número de conexiones activas es:8
2012-03 -05 13:01:41,508 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:7
…
12/03/05 13:01:41 INFO regionerver .HRegion:tabla de usuario en línea, usuario1964968041,1330944848231.dd89596e9129e1caa7e07f8a491c9734.; siguiente secuenciaid=62920
2012-03-05 13:01:41,618 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:6
2012-03-05 13 :01:41,621 DEPURACIÓN org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipc0Port=5002):El número de conexiones activas es:7
…
2012-03-05 13:01:41,829 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):El número de conexiones activas es:7
12/03/05 13:01:41 INFO regionserver.HRegion:tabla de usuario en línea ,usuario515290649,1330944849739.d23924dc9e9d5891f332c337977af83d.; siguiente secuenciaid=62926
2012-03-05 13:01:41,832 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:6
2012-03-05 13 :01:41,838 DEPURACIÓN org.apache.hadoop.hdfs.server.datanode.DataNode:DatanodeRegistration(127.0.0.1:50010, storageID=DS-1423642448-10.0.0.64-50010-1321352233772, infoPort=50075, ipc0Port=5002):El número de conexiones activas es:7
12/03/05 13:01:41 INFO regionserver.HRegion:Onlined usertable,user757669512,1330944850808.cd0d6f16d8ae9cf0c9277f5d6c6c6b9f.; siguiente secuenciaid=62929
…
2012-03-05 14:01:39,711 DEBUG org.apache.hadoop.hdfs.server.datanode.DataNode:el número de conexiones activas es:4
2012 -03-05 22:48:41,945 DEPURAR 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):El número de conexiones activas es:4
12/03/05 22:48:41 INFO regionserver.HRegion:Onlined usertable,user757669512,1330944850808.cd0d6f16d8ae9cf0c9277f5d6c6c6b9f.; siguiente secuenciaid=62929
2012-03-05 22:48:41,963 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):Número de conexiones activas:4

Puedes ver cómo las regiones se abren una tras otra, pero lo que también puedes notar es que la cantidad de conexiones activas nunca sube a 22, apenas llega a 10. ¿Porqué es eso? Para entender esto mejor, tenemos que ver cómo los archivos en HDFS se asignan a la instancia de DataXceiver del lado del servidor, y los subprocesos reales que representan.

Inmersión profunda de Hadoop

Los DFSInputStream y DFSOutputStream antes mencionados son en realidad fachadas en torno a los conceptos de flujo habituales. Envuelven la comunicación cliente-servidor en estas interfaces estándar de Java, mientras enrutan internamente el tráfico a un DataNode seleccionado, que es el que contiene una copia del bloque actual. Tiene la libertad de abrir y cerrar estas conexiones según sea necesario. Cuando un cliente lee un archivo en HDFS, las clases de la biblioteca del cliente cambian de forma transparente de bloque a bloque y, por lo tanto, de DataNode a DataNode, por lo que tiene que abrir y cerrar conexiones según sea necesario.

El DFSInputStream tiene una instancia de una clase DFSClient.BlockReader que abre la conexión al DataNode. La instancia de flujo llama a blockSeekTo() para cada llamada a read() que se encarga de abrir la conexión, si ya no hay ninguna. Una vez que un bloque se lee por completo, la conexión se cierra. Cerrar la transmisión tiene el mismo efecto, por supuesto.

El DFSOutputStream tiene una clase auxiliar similar, el DataStreamer. Realiza un seguimiento de la conexión con el servidor, que se inicia mediante el método nextBlockOutputStream(). Tiene más clases internas que ayudan a escribir los datos del bloque, que omitimos aquí por motivos de brevedad.

Tanto la escritura como la lectura de bloques requieren un subproceso para contener el socket y los datos intermedios en el lado del servidor, envueltos en la instancia de DataXceiver. Dependiendo de lo que esté haciendo su cliente, verá que la cantidad de conexiones fluctúa en torno a la cantidad de archivos a los que se accede actualmente en HDFS.

Volviendo al acertijo de HBase anterior:la razón por la que no ve hasta 22 (y más) conexiones durante el inicio es que mientras se abren las regiones, los únicos datos requeridos son el bloque de información de HFile. Este bloque se lee para obtener detalles vitales sobre cada archivo, pero luego se vuelve a cerrar. Esto significa que el recurso del lado del servidor se libera en rápida sucesión. Las cuatro conexiones restantes son más difíciles de determinar. Puede usar JStack para volcar todos los subprocesos en DataNode, que en este ejemplo muestra esta entrada:

“DataXceiver for client /127.0.0.1:64281 [sending block blk_5532741233443227208_4201]” daemon prio=5 tid=7fb96481d000 nid=0x1178b4000 ejecutable [1178b3000]
java.lang.Thread.State:EJECUTABLE
…

“DataXceiver for client /127.0.0.1:64172 [bloque de recepción blk_-2005512129579433420_4199 client=DFSClient_hb_rs_10.0.0.29 ,60020,1330984111693_1330984118810]” daemon prio=5 tid=7fb966109000 nid=0x1169cb000 ejecutable [1169ca000]
java.lang.Thread.State:RUNNABLE
…

Estas son las únicas entradas de DataXceiver (en este ejemplo), por lo que el conteo en el grupo de subprocesos es un poco engañoso. Recuerde que el subproceso del demonio DataXceiverServer ya representa una entrada adicional, que combinada con las dos anteriores representa las tres conexiones activas, lo que de hecho significa tres subprocesos activos. La razón por la que el registro indica cuatro en su lugar es que registra el recuento de un subproceso activo que está a punto de finalizar. Entonces, poco después de que se registra el conteo de cuatro, en realidad es uno menos, es decir, tres y, por lo tanto, coincide con nuestro conteo de subprocesos activos.

También tenga en cuenta que las clases auxiliares internas, como PacketResponder, ocupan otro subproceso en el grupo mientras están activas. La salida de JStack indica ese hecho, enumerando el hilo como tal:

 “PacketResponder 0 for Block blk_-2005512129579433420_4199” daemon prio=5 tid=7fb96384d000 nid=0x116ace000 en Object.wait () [116acd000]
java.lang.Thread.State:TIMED_WAITING (en el monitor de objetos)
en java.lang.Object.wait(Native Method)
en org.apache.hadoop. hdfs.server.datanode.BlockReceiver$PacketResponder \
.lastDataNodeRun(BlockReceiver.java:779)
– bloqueado (un org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder)
en org.apache.hadoop.hdfs.server.datanode.BlockReceiver$PacketResponder.run(BlockReceiver.java:870)
en java.lang.Thread.run(Thread.java:680)

Este hilo se encuentra actualmente en estado TIMED_WAITING y no se considera activo. Es por eso que el recuento emitido por las declaraciones de registro de DataXceiver no incluye este tipo de hilos. Si se activan debido a que el cliente envía datos de envío, el recuento de subprocesos activos volverá a aumentar. Otra cosa a tener en cuenta es que este subproceso no necesita una conexión separada, o socket, entre el cliente y el servidor. El PacketResponder es solo un hilo en el lado del servidor para recibir datos de bloque y transmitirlos al siguiente DataNode en la canalización de escritura.

El comando Hadoop fsck también tiene una opción para informar qué archivos están abiertos actualmente para escribir:

$ hadoop fsck /hbase -openforwrite
FSCK iniciado por larsgeorge desde /10.0.0.29 para ruta / hbase en lun 5 de marzo 22:59:47 CET 2012
……/hbase/.logs/10.0.0.29,60020,1330984111693/10.0.0.29%3A60020.1330984118842 0 bytes, 1 bloque(s), OPENFORWRITE:………………………………..Estado:SALUDABLE
Tamaño total:     2088783626 B
Total de directorios:     54
Total de archivos:   45
…

Esto no se relaciona inmediatamente con un subproceso del lado del servidor ocupado, ya que estos se asignan por ID de bloque. Pero puedes deducir de ello que hay un bloque abierto para escribir. El comando de Hadoop tiene opciones adicionales para imprimir los archivos reales y el ID de bloque que los componen:

$ hadoop fsck /hbase -files -blocks
FSCK iniciado por larsgeorge desde /10.0.0.29 para ruta /hbase el martes 6 de marzo a las 10:39:50 CET de 2012

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