Spark-on-HBase:conector HBase basado en DataFrame

Esta entrada de blog se publicó en Hortonworks.com antes de la fusión con Cloudera. Es posible que algunos enlaces, recursos o referencias ya no sean precisos.

Nos enorgullece anunciar la vista previa técnica de Spark-HBase Connector, desarrollado por Hortonworks en colaboración con Bloomberg.

El conector Spark-HBase aprovecha la API de fuente de datos (SPARK-3247) introducida en Spark-1.2.0. Cierra la brecha entre el almacén de valor clave HBase simple y las consultas SQL relacionales complejas y permite a los usuarios realizar análisis de datos complejos además de HBase usando Spark. Un HBase DataFrame es un Spark DataFrame estándar y puede interactuar con cualquier otra fuente de datos como Hive, ORC, Parquet, JSON, etc.

Antecedentes

Hay varios conectores Spark HBase de código abierto disponibles como paquetes Spark, como proyectos independientes o en troncal HBase.

Spark se ha movido a las API Dataset/DataFrame, que proporciona una optimización del plan de consulta integrado. Ahora, los usuarios finales prefieren usar la interfaz basada en DataFrames/Datasets.

El conector HBase en el enlace troncal HBase tiene un gran soporte a nivel de RDD, p. BulkPut, etc., pero su compatibilidad con DataFrame no es tan rica. El conector troncal de HBase se basa en HadoopRDD estándar con TableInputFormat integrado de HBase y tiene algunas limitaciones de rendimiento. Además, BulkGet realizado en el controlador puede ser un único punto de error.

Hay algunas otras implementaciones alternativas. Tome Spark-SQL-on-HBase como ejemplo. Aplica técnicas de optimización personalizadas muy avanzadas mediante la incorporación de su propio plan de optimización de consultas dentro del motor Spark Catalyst estándar, envía el RDD a HBase y realiza tareas complicadas, como la agregación parcial, dentro del coprocesador HBase. Este enfoque es capaz de lograr un alto rendimiento, pero es difícil de mantener debido a su complejidad y la rápida evolución de Spark. Además, permitir que se ejecute código arbitrario dentro de un coprocesador puede plantear riesgos de seguridad.

El conector Spark-on-HBase (SHC) se ha desarrollado para superar estos cuellos de botella y debilidades potenciales. Implementa la API Spark Datasource estándar y aprovecha el motor Spark Catalyst para la optimización de consultas. Paralelamente, el RDD se construye desde cero en lugar de usar TableInputFormat para lograr un alto rendimiento. Con este RDD personalizado, todas las técnicas críticas se pueden aplicar e implementar por completo, como la eliminación de particiones, la eliminación de columnas, la inserción de predicados y la localidad de datos. El diseño hace que el mantenimiento sea muy fácil, al tiempo que logra un buen compromiso entre rendimiento y simplicidad.

Arquitectura

Asumimos que Spark y HBase se implementan en el mismo clúster y que los ejecutores de Spark se ubican junto con los servidores regionales, como se ilustra en la figura a continuación.

Figura 1. Arquitectura del conector Spark-on-HBase

En un nivel alto, el conector trata a Scan y Get de manera similar, y ambas acciones se realizan en los ejecutores. El controlador procesa la consulta, agrega exploraciones/obtenciones en función de los metadatos de la región y genera tareas por región. Las tareas se envían a los ejecutores preferidos ubicados junto con el servidor de la región y se realizan en paralelo en los ejecutores para lograr una mejor localidad y concurrencia de datos. Si una región no contiene los datos necesarios, a ese servidor de región no se le asigna ninguna tarea. Una tarea puede constar de varios escaneos y BulkGets, y las solicitudes de datos de una tarea se recuperan de un solo servidor de región, y este servidor de región también será la preferencia de localidad para la tarea. Tenga en cuenta que el controlador no está involucrado en la ejecución del trabajo real, excepto en las tareas de programación. Esto evita que el conductor sea el cuello de botella.

Catálogo de mesas

Para traer la tabla HBase como una tabla relacional a Spark, definimos un mapeo entre las tablas HBase y Spark, llamado Table Catalog. Hay dos partes críticas de este catálogo. Una es la definición de clave de fila y la otra es la asignación entre la columna de la tabla en Spark y la familia de columnas y el calificador de columna en HBase. Consulte la sección Uso para obtener más detalles.

Soporte nativo de Avro

El conector admite el formato Avro de forma nativa, ya que es una práctica muy común conservar los datos estructurados en HBase como una matriz de bytes. El usuario puede conservar el registro de Avro en HBase directamente. Internamente, el esquema de Avro se convierte automáticamente a un tipo de datos nativo de Spark Catalyst. Tenga en cuenta que ambas partes clave-valor en una tabla HBase se pueden definir en formato Avro. Consulte los ejemplos/casos de prueba en el repositorio para conocer el uso exacto.

Inserción de predicado

El conector solo recupera las columnas requeridas del servidor de la región para reducir la sobrecarga de la red y evitar el procesamiento redundante en el motor Spark Catalyst. Los filtros HBase estándar existentes se utilizan para realizar la inserción de predicados sin aprovechar la capacidad del coprocesador. Debido a que HBase no conoce el tipo de datos, excepto la matriz de bytes, y la inconsistencia de orden entre los tipos primitivos de Java y la matriz de bytes, tenemos que procesar previamente la condición del filtro antes de configurar el filtro en la operación de escaneo para evitar cualquier pérdida de datos. Dentro del servidor de región, se filtran los registros que no coinciden con la condición de consulta.

Recorte de particiones

Al extraer la clave de fila de los predicados, dividimos Scan/BulkGet en múltiples rangos que no se superponen, solo los servidores de región que tienen los datos solicitados realizarán Scan/BulkGet. Actualmente, la poda de partición se realiza en la primera dimensión de las claves de fila. Por ejemplo, si una clave de fila es "clave1:clave2:clave3", la eliminación de la partición se basará únicamente en "clave1". Tenga en cuenta que las condiciones DONDE deben definirse cuidadosamente. De lo contrario, es posible que la eliminación de la partición no surta efecto. Por ejemplo, WHERE filaclave1> “abc” O columna =“xyz” (donde filaclave1 es la primera dimensión de la clave de fila y la columna es una columna normal de hbase) dará como resultado un análisis completo, ya que tenemos que cubrir todos los rangos porque del O lógica.

Localidad de datos

Cuando un ejecutor de Spark se ubica junto con los servidores de la región HBase, la ubicación de los datos se logra mediante la identificación de la ubicación del servidor de la región y se hace el mejor esfuerzo para ubicar la tarea junto con el servidor de la región. Cada ejecutor realiza Scan/BulkGet en la parte de los datos coubicados en el mismo host.

Escanear y obtener masivamente

Estos dos operadores están expuestos a los usuarios especificando la CLÁUSULA WHERE, por ejemplo, WHERE columna> x y columna para escanear y DONDE columna =x olvidar. Las operaciones se realizan en los ejecutores, y el controlador solo construye estas operaciones. Internamente, se convierten para escanear y/u obtener, e Iterator[Row] se devuelve al motor de Catalyst para el procesamiento de la capa superior.

[code language="scala"]def catalog = s"""{
         |"table":{"namespace":"default", "name":"table1"},
         |"rowkey":"key",
         |"columns":{
           |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
           |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
           |"col2":{"cf":"cf2", "col":"col2", "type":"double"},
           |"col3":{"cf":"cf3", "col":"col3", "type":"float"},
           |"col4":{"cf":"cf4", "col":"col4", "type":"int"},
           |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
           |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
           |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
           |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
         |}
       |}""".stripMargin
[/code]