Nos emociona anunciar la Disponibilidad General (GA) del soporte de Python para Conector de Couchbase para Spark, que ofrece una integración de primera clase entre Couchbase Server y Apache Spark para los ingenieros de datos de Python. Esta versión GA significa que el conector está listo para producción y cuenta con soporte completo, lo que permite que las aplicaciones PySpark lean y escriban en Couchbase sin problemas. Con la base de datos NoSQL de alto rendimiento de Couchbase (con el lenguaje de consultas SQL++/SQL++) y el motor de procesamiento distribuido de Spark, los ingenieros de datos ahora pueden combinar fácilmente estas tecnologías para construir canales de datos y flujos de trabajo de análisis rápidos y escalables. En resumen, el conector de Couchbase Spark para PySpark desbloquea una integración de datos eficiente y en paralelo, lo que le permite aprovechar Spark para ETL/ELT, análisis en tiempo real, aprendizaje automático y más en los datos almacenados en Couchbase.
En esta publicación, cubriremos cómo comenzar con el conector de PySpark, demostraremos operaciones básicas de lectura y escritura (tanto basadas en clave-valor como en consultas) para la base de datos operacional de Couchbase y las bases de datos columnares de Capella; y compartiremos consejos de optimización del rendimiento para obtener el mejor rendimiento. Ya sea que haya estado usando el conector de Couchbase para Spark en Scala, o si es nuevo en la integración entre Couchbase y Spark, esta guía lo ayudará a ponerse al día rápidamente usando PySpark para sus necesidades de ingeniería de datos.
¿Por qué PySpark?
La incorporación de soporte para PySpark en el conector existente de Couchbase para Spark estuvo impulsada por la creciente demanda de ingenieros de datos y desarrolladores que prefieren Python por su simplicidad y su enorme ecosistema de aprendizaje automático en Python para Spark en flujos de trabajo de ingeniería y ciencia de datos. Este soporte garantiza que los equipos que ya usan Python ahora puedan integrar Couchbase (ya sea que estén usando Couchbase Capella (DBaaS), base de datos operacional autogestionada o Columna Capella base de datos) a flujos de trabajo de Spark basados en Python, lo que permite una adopción más amplia y procesos de datos optimizados.
La supremacía de Python en los casos de uso de IA/ML, respaldada por marcos de trabajo como SparkML, PyTorch, TensorFlow, H2O, DataRobot, scikit-learn y SageMaker, así como por populares herramientas de análisis exploratorio de datos como Matplotlib y Plotly, subraya aún más la necesidad de la integración con PySpark. Además, la compatibilidad con PySpark desbloquea tuberías de ETL y ML aceleradas mediante el uso de aceleración por GPU (Spark RAPIDS) y facilita tareas sofisticadas de ingeniería de características y manipulación de datos utilizando bibliotecas ampliamente adoptadas como Pandas, NumPy y las API de ingeniería de características integradas de Spark. Este nuevo soporte agiliza significativamente los procesos de datos y amplía las oportunidades de adopción para Couchbase en los equipos de ciencia e ingeniería de datos.
Primeros pasos con Couchbase PySpark
Empezar es sencillo. El conector de Couchbase para Spark se distribuye como un único archivo JAR (archivo Java) que se agrega al entorno de Spark. Puede obtener el conector desde el sitio oficial Sitio de descargas de Couchbase o a través de Coordenadas de Maven. Una vez que tengas el JAR, usarlo en PySpark es tan sencillo como configurar tu sesión de Spark con el conector y los parámetros de conexión de Couchbase.
1. Obtenga o cree una base de datos operacional de Couchbase o una base de datos Capella Columnar
La forma más rápida de empezar con Couchbase es utilizar nuestro Base de datos como servicio Capella. Una vez allí, puedes buscar tu base de datos existente o crear una operativo o en columnas (para análisis) base de datos. Como alternativa, puedes usar nuestro Couchbase autogestionado.
2. Instalar PySpark (si no está instalado)
Si estás trabajando en un entorno de Python, instala PySpark usando pip. Por ejemplo, en un entorno virtual:
|
1 |
pip Instalar PySpark |
Esto instalará Apache Spark para su uso con Python. Si estás ejecutando en un clúster de Spark existente o en Databricks, es posible que PySpark ya esté disponible.
3. Incluya el JAR del conector de Couchbase para Spark
Descargar el spark-connector-assembly-.jar para la última versión del conector. Luego, al crear su sesión de Spark o enviar su trabajo, proporcione este JAR en la configuración. Puede hacer esto configurando el --jars opción en spark-submit o mediante el generador de SparkSession en el código (como se muestra a continuación).
4. Configure la conexión de Couchbase
Debe especificar la cadena de conexión del clúster de Couchbase y las credenciales (nombre de usuario y contraseña). En Capella, puede encontrar esto en la pestaña “Conectar” para operativo y Configuración->Cadena de conexión para columnar. Opcionalmente, especifique un bucket o scope predeterminado si es necesario (aunque también puede especificar el bucket o scope por operación).
A continuación se muestra ejemplo rápido de PySpark eso establece un SparkSession para conectarse a un clúster de Couchbase y luego leer algunos datos:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
desde PySpark.SQL importar SparkSession # Inicializar SparkSession con el conector de Couchbase y la configuración de conexión chispa = SparkSession.constructor .nombre de la aplicación(“CouchbaseIntegrationExample”) .maestro(“local[*]”) # utilizando Spark local, por ejemplo; omítelo o adáptalo para un clúster de Spark .configuración(“spark.jars”, “/ruta/a/spark-connector-assembly-.jar””) .configuración(“spark.couchbase.connectionString”, “couchbases://”) .configuración(“spark.couchbase.username”, “””) .configuración(“spark.couchbase.password”, “””) .obtener u crear() # Prueba la conexión leyendo algunos documentos de Couchbase (utilizando un bucket de ejemplo) df = chispa.leer.formato(“couchbase.query”) .opción(“cubeta”, “nombre_del_bucket”) .opción(“alcance”, “scope_name”) .opción(“colección”, “nombre_de_colección”) .cargar() df.imprimirEsquema() df.mostrar(5) |
En el código anterior, configuramos la sesión de Spark para incluir el JAR del conector de Couchbase y la apuntamos a un clúster de Couchbase. Luego creamos un DataFrame df leyendo desde el nombre_del_bucket cubocope_name.collection_name colección) a través del servicio Query.
Por el resto de este documento, asumimos que ha cargado nuestro conjunto de datos de muestra travel-sample qué se puede hacer para Couchbase Capella operacional o En columna muy fácilmente.
Leer y escribir en Couchbase utilizando PySpark
Una vez que su sesión de Spark esté conectada a Couchbase, podrá realizar ambas acciones: Operaciones clave-valor (para escritos) y Operaciones de consulta (usando SQL++ tanto para la lectura como para la escritura) a través de DataFrames.
La siguiente tabla muestra el formato que el conector Sparks admite para leer y escribir en las bases de datos Couchbase y columnar:
| Base de datos operativa de Couchbase/Capella | Base de datos Capella Columnar | |
| Operaciones de lectura | Leer.format("couchbase.query") |
leer.format("couchbase.columnar") |
| Operaciones de escritura | (recomendado para usar el servicio de datos)
|
escritura.format("couchbase.columnar") |
Lectura de Couchbase con un DataFrame de consultas
El conector Couchbase Spark permite cargar datos desde un bucket de Couchbase como un DataFrame Spark mediante consultas SQL++. Al utilizar el lector de DataFrame con formato couchbase.query, Con esta opción, puede especificar un bucket (y un ámbito/colección) y parámetros de consulta opcionales. Por ejemplo, para leer todos los documentos de una colección o de un subconjunto definido por un filtro:
|
1 2 3 4 5 6 7 8 9 10 |
# Lea todos los documentos de una colección de Couchbase utilizando el servicio de consulta airlines_df = chispa.leer.formato(“couchbase.query”) .opción(“cubeta”, “travel-sample”) .opción(“alcance”, “inventario”) .opción(“colección”, “aerolínea”) .cargar() # Ejemplo: filtrar el DataFrame utilizando Spark (se enviará a Couchbase siempre que sea posible) usa_airlines_df = airlines_df.filtro(“country = ‘Estados Unidos'”) usa_airlines_df.mostrar(5) |
En este ejemplo, airlines_df carga todos los documentos desde el travel-sample.inventory.airline La recopilación se realiza en un Spark DataFrame. Luego aplicamos un filtro para encontrar aerolíneas con sede en Estados Unidos. El conector intentará empujar hacia abajo filtros para Couchbase para que no se transfiera datos innecesarios (es decir, incluirá el WHERE country = 'Estados Unidos' la cláusula de la consulta de SQL++ que se ejecuta, si es posible). El resultado, usa_airlines_df, se puede usar como cualquier otro DataFrame en Spark (por ejemplo, se puede combinar con otros DataFrames, aplicar agregaciones, etc.).
En su interior, el Particiones de conectores El resultado de la consulta genera múltiples tareas si está configurado (más sobre esto en Optimización de rendimiento abajo), y utiliza el servicio de consultas de Couchbase (alimentado por el motor SQL++) para obtener los datos. Cada partición de Spark corresponde a un subconjunto de datos obtenido mediante una consulta equivalente de SQL++. Esto permite lecturas paralelas desde Couchbase, aprovechando la naturaleza distribuida tanto de Spark como de Couchbase.
Escribir a Couchbase mediante operaciones de clave-valor (KV) (recomendado)
El conector también admite la escritura de datos a Couchbase, ya sea mediante la Servicio de datos (KV) o a través del servicio de consulta (ejecución de SQL++ INSERT/UPSERT instrucciones para usted). El recomendado La forma más común de utilizarlo es usar la Fuente de datos clave-valor (format("couchbase.kv")) para Mejor rendimiento. En el modo clave-valor, cada tarea de Spark escribirá documentos directamente en los nodos de datos de Couchbase.
Al escribir un DataFrame en Couchbase, debe asegurarse de que cada documento tenga un ID único (ya que Couchbase requiere un ID de documento). Por defecto, el conector busca una columna llamada __META_ID (o META_ID en versiones más recientes) en el DataFrame para el ID del documento. También puede especificar un campo de ID personalizado mediante el parámetro IdFieldName opción.
Supongamos, por ejemplo, que tenemos un DataFrame de Spark new_airlines_df que queremos escribir a Couchbase. Tiene una columna airline_id Esto debería servir como clave del documento de Couchbase, y el resto de las columnas son el contenido del documento:
|
1 2 3 4 5 6 7 8 |
# Supongamos que new_airlines_df es un DataFrame al que queremos escribir en Couchbase. # Contiene una columna “airline_id” que se utilizará como ID del documento. new_airlines_df.escrever.formato(“couchbase.kv”) .opción(“cubeta”, “mybucket”) .opción(“alcance”, “myscope”) .opción(“colección”, “aerolíneas”) .opción(“idFieldName”, “airline_id”) .guardar() |
Escribiendo a Couchbase mediante operaciones de Query (SQL++)
Si bien recomendamos usar el servicio de datos (KV) como se describe anteriormente, ya que suele ser más rápido que el servicio de consultas, también puede escribir mediante el servicio de consultas utilizando format("couchbase.query") en write. Esto ejecutará internamente las instrucciones SQL++ UPSERT para cada fila. Esto puede ser útil si necesitas aprovechar una función de SQL++ (por ejemplo, transformaciones en el lado del servidor), pero para insertaciones o actualizaciones simples, el enfoque de KV es más eficiente.
|
1 2 3 4 5 6 |
df.escrever.formato(“couchbase.query”) .opción(“cubeta”, “mybucket”) .opción(“alcance”, “myscope”) .opción(“colección”, “aerolíneas”) .modo(“sobreescriber”) .guardar() |
En la siguiente sección, modificaremos estos casos básicos de lectura y escritura para el último producto de análisis de Couchbase, Capella Columnar.
Compatibilidad de PySpark con la columna Capella
Una de las nuevas características clave del conector Couchbase Spark GA es el soporte para Capella Columnar. Capella Columnar es un servicio de base de datos analítica nativa en Couchbase Capella que almacena datos en un formato orientado a columnas para análisis de alto rendimiento.
Lectura de datos formateados en columna con PySpark
La lectura de datos desde un clúster columnar de Couchbase Capella en PySpark es similar al clúster operativo de Couchbase, con tres diferencias:
- Usa
format("couchbase.columnar")La opción de especificar conexión es para el servicio columnar. - La cadena de conexión para el tipo de columna se puede obtener a través de la interfaz de usuario de Capella.
- También especificará qué conjunto de datos cargará proporcionando los nombres de la base de datos, el ámbito y la colección (análogo a los buckets/scopes/collections en Couchbase) como opciones.
Una vez configurado Spark, puede usar la API de lector de DataFrame de Spark para cargar datos desde el servicio columnar:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
desde PySpark.SQL importar SparkSession # Inicializar SparkSession con configuraciones de Couchbase (si el jar del conector está disponible) chispa = SparkSession.constructor .nombre de la aplicación(“Ejemplo de columna de Couchbase Spark Connector”) .configuración(“spark.couchbase.connectionString”, “couchbases://your.columnar.connection.string”) .configuración(“spark.couchbase.username”, “Tu nombre de usuario en columna”) .configuración(“spark.couchbase.password”, “YourColumnarPassword”) .obtener u crear() # Lea un DataFrame desde la colección de columnas de Couchbase Capella (collección travel-sample.inventory.airline) airlines_df = chispa.leer.formato(“couchbase.columnar”) .opción(“base de datos”, “travel-sample”) .opción(“alcance”, “inventario”) .opción(“colección”, “aerolínea”) .cargar() |
En este ejemplo, el resultado airlines_df Es un Spark DataFrame normal — puedes inspeccionarlo, ejecutar transformaciones y realizar acciones como .count() o .show() Como siempre. Por ejemplo, airlines_df.show(5) imprimirá algunos documentos de la aerolínea, y airlines_df.count() Devolverá el número de documentos de la colección. En la parte interna, el conector deduce automáticamente un esquema para los documentos JSON mediante la obtención de muestras de hasta un cierto número de registros (por defecto 1000). Todos los campos que aparecen de forma consistente en los documentos muestreados se convierten en columnas en el DataFrame, con los tipos de datos apropiados de Spark.
Tenga en cuenta que si sus documentos tienen esquemas diferentes, la inferencia podría producir un esquema que incluya la unión de todos los campos (los campos que no estén presentes en algunos documentos serán nulos en esas filas). En los casos en que el esquema esté evolucionando o desee restringir qué registros se consideran, puede proporcionar un filtro (predicado) explícito al lector, como se describe a continuación.
Consultar un conjunto de datos columnar en Couchbase mediante Spark
A menudo puede que no desee cargar toda la colección, especialmente si esta es muy grande. Puede optimizar el rendimiento al desplazar los predicados de filtrado directamente al servicio Capella Columnar al cargar los datos, evitando así la transferencia de datos innecesaria. .option("filter", "") aplicar una cláusula WHERE de SQL++ durante la operación de lectura. Por ejemplo, para cargar solo aerolíneas con sede en Estados Unidos:
|
1 2 3 4 5 6 7 8 |
usa_airlines_df = chispa.leer.formato(“couchbase.columnar”) .opción(“base de datos”, “travel-sample”) .opción(“alcance”, “inventario”) .opción(“colección”, “aerolínea”) .opción(“filtro”, “country = ‘Estados Unidos'”) .cargar() imprimir(usa_airlines_df.contador()) # Solo las aerolíneas donde el país = ‘Estados Unidos’ |
El connector ejecuta este filtro directamente en la fuente, recuperando solo los documentos relevantes. También es posible desplazar las proyecciones (seleccionando campos específicos) y las agregaciones en algunos casos – el connector se encargará de ello. Agregados simples como CONTAR, MIN, MAX, y SUM Al motor Columnar siempre que sea posible, en lugar de calcularlos en Spark, para obtener mejores resultados.
Una vez que los datos se cargan en un DataFrame, se pueden realizar operaciones estándar Desarrolla transformaciones, agregaciones y agrupaciones. Por ejemplo, para contar las aerolíneas por país utilizando Spark SQL, incluso se puede crear una vista temporal para ejecutar consultas de Spark SQL sobre los datos de la siguiente manera:
|
1 2 3 4 5 6 7 8 |
airlines_df.createOrReplaceTempView(“airlines_view”) result_df = chispa.SQL(“”“ SELECT país, COUNT(*) como número de aerolíneas Desde airlines_view AGRUPAR POR país ORDENAR POR número de vuelos de aerolínea DESC ““”) result_df.mostrar(10) |
Esta consulta se ejecuta completamente dentro del motor de Spark, lo que brinda flexibilidad para integrar de forma perfecta los datos de Couchbase en flujos de trabajo analíticos complejos.
Tras haber abordado conceptos básicos sobre leer y escribir datos, pasemos a ver cómo podemos optimizar el rendimiento al mover grandes volúmenes de datos entre Couchbase y Spark.
Consejos para la optimización de rendimiento
Para maximizar el rendimiento y la eficiencia al utilizar el conector Couchbase PySpark, considere las siguientes mejores prácticas.
Ajustar las operaciones de lectura
Utilice la segmentación de consultas para lograr un paralelismo
(Couchbase Capella (DBaaS), base de datos operacional autogestionada o Columna Capella)
Al leer a través del servicio de consulta de bases de datos operativas o columnar, aprovecha la capacidad del conector para dividir los resultados de la consulta. Puede especificar una partitionCount (y un campo de partición numérico con límites superior e inferior) para la lectura del DataFrame. Una buena regla general es establecer partitionCount a al menos el número total de núcleos de CPU del servicio de consultas disponible en su clúster de Couchbase. Esto garantiza que Spark ejecute múltiples consultas en paralelo, aprovechando todos los nodos de consulta. Por ejemplo, si el servicio de consultas de su clúster de Couchbase tiene un total de 8 núcleos, configure partitionCount >= 8 de modo que se generen al menos 8 consultas SQL++ paralelas. Esto puede aumentar drásticamente el rendimiento de lectura al utilizar todos los nodos de consulta simultáneamente. Tenga en cuenta que también debe haber suficientes núcleos en su clúster de Spark para ejecutar tantas consultas paralelas.
El uso de índices de cobertura para la eficiencia de la consulta
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
Si utiliza consultas SQL++, intente realizar consultas mediante Cobertura de índices Siempre que sea posible. Un índice de cobertura es un índice que incluye todo campos que necesita la consulta, de modo que la consulta pueda ser servida completamente desde el índice sin tener que obtener los datos del servicio de datos. Las consultas cubiertas evitan el paso adicional por la red para obtener los documentos completos, lo que permite: proporcionando un rendimiento mejoradoDiseña los índices secundarios de Couchbase para incluir los campos que filtras y Los campos que devuelva, si es posible. Esto podría implicar crear índices específicos para sus trabajos de Spark que cubran exactamente los datos necesarios.
Asigne réplicas de índice para evitar colas
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
Además de utilizar índices de cobertura, asegúrate de que los índices se repliquen en múltiples nodos de índice. Replicación de índices no solo proporciona alta disponibilidad, sino que también permite que las consultas se realicen carga equilibrada entre las copias del índice en diferentes nodos para obtener una mayor eficienciaEn la práctica, si dispones de (por ejemplo) 3 nodos de índice, replicar los índices importantes entre ellos significa que las consultas paralelas del conector Spark pueden llegar a diferentes nodos de índice en lugar de que todos los procesos se concentren en un único nodo.
Ajustar las operaciones de escritura
Prefiero el servicio de datos para las escrituras masivas
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
Recomendamos utilizar la fuente de datos clave-valor (Servicio de datos) en lugar del servicio de consultas para operaciones de escritura. La escritura a través del servicio de datos (actualizaciones directas de KV) suele ser varias veces más rápido en comparación con las inserciones basadas en SQL++. De hecho, los análisis internos han demostrado que la escritura mediante KV puede ser aproximadamente 3x más rápido en lugar de usar SQL++ en los trabajos de Spark. Esto se debe a que el servicio de datos puede ingerir documentos en paralelo directamente a los nodos responsables, con una latencia más baja por operación. Tenga en cuenta que los índices se actualizan por separado, si es necesario, para esos nuevos documentos, ya que los flujos de escritura KV no activarán automáticamente las actualizaciones de índice más allá del índice principal.
Aumentar la cantidad de particiones de escritura para los escritos del servicio de consultas
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
Aunque no se recomienda, si decides utilizar couchbase.query para la escritura (por ejemplo, si se realizan transformaciones en el lado del servidor mientras se escribe), optimice el rendimiento utilizando un alto número de particiones de escritura. Puede reparticionar su DataFrame antes de escribir para que Spark ejecute muchas tareas de escritura simultáneas. Una guía aproximada es usar en el orden de cientos de particiones Para escritas a gran escala mediante SQL++. Por ejemplo, utilizando aproximadamente 128 particiones por nodo de consulta CPU Es un punto de partida que algunos usuarios han encontrado eficaz. Esto significa que si tienes 8 núcleos de consulta, prueba con ~1024 particiones. La idea es inundar el servicio de consultas con suficientes sentencias UPSERT paralelas para maximizar el rendimiento. Sea cauteloso y encuentre el equilibrio adecuado para su clúster: una concurrencia demasiado alta podría sobrecargar el servicio de consultas. Mida el rendimiento de consulta de Couchbase y ajuste en consecuencia.
Siguiendo estos consejos de ajuste de parámetros –alineando los números de particiones con los recursos del clúster, indexando de forma inteligente y eligiendo el servicio adecuado para el trabajo– se puede lograr un rendimiento óptimo para la integración de Couchbase-Spark. Mantenga un ojo tanto en las métricas de trabajo de Spark como en las estadísticas de rendimiento de Couchbase (disponibles en la interfaz de usuario y en los registros de Couchbase) para identificar cualquier cuello de botella (por ejemplo, si un nodo de consulta está realizando toda la tarea o si la red está saturada) y ajustar la configuración según sea necesario.
Comunidad y apoyo
El soporte de Couchbase PySpark se basa en el Conector de Couchbase Spark para Couchbase y es de código abierto, y te animamos a que participes, nos des tu opinión y se unas a la conversación. puedes acceder a nuestro completo documentación, ¡Únete a nosotros Foros de Couchbase o Discord de Couchbase.
Para más información, consulte:
Para obtener más información y documentación detallada, consulte el documento oficial. Documentación del conector Couchbase Spark y la sección relevante sobre PySpark:
- Documentación de Couchbase PySpark
- Repositorio de GitHub del conector Couchbase Spark
- Foros de Couchbase (sección de Spark Connector)
- Ejemplo de la conexión Couchbase PySpark con Jupyter Notebook
- Ejemplo de ML de Couchbase PySpark en una nota de Jupyter: Cancelaciones de hoteles
¡Buena programación!
El equipo de Couchbase
Deja un comentario
Lo siento, debes estar conectado para publicar un comentario.