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.
Read/write to Couchbase using PySpark
Once your Spark session is connected to Couchbase, you can perform both key-value operations (for writes) and query operations (using SQL++ for both read and writes) through DataFrames.
Following table shows the format Sparks connector supports to read and write to Couchbase and columnar databases:
| Couchbase/Capella operational database | Capella Columnar database | |
| Read operations | read.format("couchbase.query") |
read.format("couchbase.columnar") |
| Write operations | (recommended to use Data Service)
|
write.format("couchbase.columnar") |
Reading from Couchbase with a Query DataFrame
The Couchbase Spark Connector allows you to load data from a Couchbase bucket as a Spark DataFrame via SQL++ queries. Using the DataFrame reader with format couchbase.query, you can specify a bucket (and scope/collection) and optional query parameters. For example, to read all documents from a collection or a subset defined by a filter:
|
1 2 3 4 5 6 7 8 9 10 |
# Read all documents from a Couchbase collection using the Query service airlines_df = chispa.leer.formato(“couchbase.query”) .opción(“cubeta”, “travel-sample”) .opción(“alcance”, “inventory”) .opción(“colección”, “airline”) .cargar() # Example: filter the DataFrame using Spark (will push down to Couchbase where possible) usa_airlines_df = airlines_df.filter(“country = ‘United States'”) usa_airlines_df.mostrar(5) |
In this example, airlines_df loads all documents from the travel-sample.inventory.airline collection into a Spark DataFrame. We then apply a filter to find airlines based in the United States. The connector will attempt to push down filters to Couchbase so that unnecessary data isn’t transferred (i.e. it will include the WHERE country = 'United States' clause in the SQL++ query it runs, if possible). The result, usa_airlines_df, can be used like any other DataFrame in Spark (for example, you could join it with other DataFrames, apply aggregations, etc.).
Under the hood, the connector partitions the query results into multiple tasks if configured (more on this in Performance Tuning below), and uses Couchbase’s Query service (powered by the SQL++ engine) to retrieve the data. Each Spark partition corresponds to a subset of data retrieved by an equivalent SQL++ query. This allows parallel reads from Couchbase, leveraging the distributed nature of both Spark and Couchbase.
Writing to Couchbase with Key-Value (KV) operations (recommended)
The connector also supports writing data to Couchbase, either via the Data service (KV) or via the Query service (executing SQL++ INSERT/UPSERT commands for you). The recommended way for most use cases is to use the Key-Value data source (format("couchbase.kv")) for better performance. In key-value mode, each Spark task will write documents directly to Couchbase data nodes.
When writing a DataFrame to Couchbase, you must ensure there is a unique ID for each document (since Couchbase requires a document ID). By default, the connector looks for a column named __META_ID (or META_ID in newer versions) in the DataFrame for the document ID. You can also specify a custom ID field via the IdFieldName option.
For example, suppose we have a Spark DataFrame new_airlines_df that we want to write to Couchbase. It has a column airline_id that should serve as the Couchbase document key, and the rest of the columns are the document content:
|
1 2 3 4 5 6 7 8 |
# Assume new_airlines_df is a DataFrame we want to write to Couchbase # It contains an “airline_id” column to use as the document ID. new_airlines_df.write.formato(“couchbase.kv”) .opción(“cubeta”, “mybucket”) .opción(“alcance”, “myscope”) .opción(“colección”, “airlines”) .opción(“idFieldName”, “airline_id”) .save() |
Writing to Couchbase with Query (SQL++) operations
While we recommend using the Data service (KV) as above as it is typically faster than Query service, if you prefer, you can also write via the Query service by using format("couchbase.query") on write. This will internally execute SQL++ UPSERT statements for each row. This may be useful if you need to leverage a SQL++ feature (for example, server-side transformations), but for straightforward inserts/updates, the KV approach is more efficient.
|
1 2 3 4 5 6 |
df.write.formato(“couchbase.query”) .opción(“cubeta”, “mybucket”) .opción(“alcance”, “myscope”) .opción(“colección”, “airlines”) .mode(“overwrite”) .save() |
In the next section, let us modify these basic read/write cases for Couchbase’s latest analytics product – Capella Columnar.
PySpark support for Capella Columnar
One of the key new features in the Couchbase Spark Connector GA is Capella Columnar support. Capella Columnar is a JSON-native analytical database service in Couchbase Capella that stores data in a column-oriented format for high-performance analytics
Reading Columnar-Formatted Data with PySpark
Reading data from a Couchbase Capella Columnar cluster in PySpark is similar to couchbase operational cluster except three changes:
- Use the
format("couchbase.columnar")to specify connection is for columnar service. - The connection string for columnar can be retrieved from Capella UI.
- You will also specify which dataset to load by providing the database, scope, and collection names (analogous to bucket/scope/collection in Couchbase) as options
Once Spark is configured, you can use the Spark DataFrame reader API to load data from the columnar service:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
desde PySpark.SQL importar SparkSession # Initialize SparkSession with Couchbase configs (assuming connector jar is available) chispa = SparkSession.constructor .nombre de la aplicación(“Couchbase Spark Connector Columnar Example”) .configuración(“spark.couchbase.connectionString”, “couchbases://your.columnar.connection.string”) .configuración(“spark.couchbase.username”, “YourColumnarUsername”) .configuración(“spark.couchbase.password”, “YourColumnarPassword”) .obtener u crear() # Read a DataFrame from Couchbase Capella Columnar (travel-sample.inventory.airline collection) airlines_df = chispa.leer.formato(“couchbase.columnar”) .opción(“database”, “travel-sample”) .opción(“alcance”, “inventory”) .opción(“colección”, “airline”) .cargar() |
In this example, the resulting airlines_df is a normal Spark DataFrame — you can inspect it, run transformations, and perform actions like .count() o .show() as usual. For instance, airlines_df.show(5) will print a few airline documents, and airlines_df.count() will return the number of documents in the collection. Under the hood, the connector automatically infers a schema for the JSON documents by sampling up to a certain number of records (by default 1000). All fields that consistently appear in the sampled documents become columns in the DataFrame, with appropriate Spark data types.
Note that if your documents have varying schemas, the inference might produce a schema that includes the union of all fields (fields not present in some documents will be null in those rows). In cases where the schema is evolving or you want to restrict which records are considered, you can provide an explicit filter (predicate) to the reader, as described next.
Querying a Columnar Dataset in Couchbase via Spark
Often you may not want to load an entire collection, especially if it’s large. You can optimize performance by pushing down filter predicates directly to the Capella Columnar service when loading data, avoiding unnecessary data transfer. Use .option("filter", "") to apply a SQL++ WHERE clause during the read operation. For instance, to load only airlines based in the United States:
|
1 2 3 4 5 6 7 8 |
usa_airlines_df = chispa.leer.formato(“couchbase.columnar”) .opción(“database”, “travel-sample”) .opción(“alcance”, “inventory”) .opción(“colección”, “airline”) .opción(“filter”, “country = ‘United States'”) .cargar() print(usa_airlines_df.count()) # Only airlines where country = ‘United States’ |
The connector executes this filter directly at the source, retrieving only relevant documents. You can also push down projections (selecting specific fields) and aggregations in some cases – the connector will offload simple aggregates like COUNT, MIN, MAX, y SUM to the Columnar engine whenever possible, rather than computing them in Spark, for better performance
Once data is loaded into a DataFrame, you can perform standard Spark transformations, joins, and aggregations. For example, to count airlines per country using Spark SQL, you can even create a temporary view to run Spark SQL queries on the data as follows:
|
1 2 3 4 5 6 7 8 |
airlines_df.createOrReplaceTempView(“airlines_view”) result_df = chispa.SQL(“”“ SELECT country, COUNT(*) AS airline_count FROM airlines_view GROUP BY country ORDER BY airline_count DESC ““”) result_df.mostrar(10) |
This query runs entirely within Spark engine, giving flexibility to integrate Couchbase data seamlessly into complex analytical workflows.
Having covered basic reads and writes, let’s move on to how you can tune performance when moving large volumes of data between Couchbase and Spark.
Performance tuning tips
To maximize throughput and efficiency when using the Couchbase PySpark Connector, consider the following best practices.
Tuning your read operations
Use Query Partitioning for parallelism
(Couchbase Capella (DBaaS), base de datos operacional autogestionada o Columna Capella)
When reading via the Query service for operational or columnar database, take advantage of the connector’s ability to partition the query results. You can specify a partitionCount (and a numeric partitioning field with lower/upper bounds) for the DataFrame read. A good rule of thumb is to set partitionCount to at least the total number of query service CPU cores available in your Couchbase cluster. This ensures Spark will run multiple queries in parallel, leveraging all query nodes. For example, if your Couchbase cluster’s Query service has 8 cores in total, set partitionCount >= 8 so that at least 8 parallel SQL++ queries will be issued. This can dramatically increase read throughput by utilizing all query nodes concurrently. Note that you must have enough cores in your Spark cluster as well to run that many parallel queries.
Leverage covering indexes for query efficiency
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
If using SQL++ queries, try to query through covering indexes whenever possible. A covering index is an index that includes todo fields your query needs, so the query can be served entirely from the index without fetching from the data service. Covered queries avoid the extra network hop to fetch full documents, thus delivering better performance. Design your Couchbase secondary indexes to include the fields you filter on y the fields you return, if feasible. This might mean creating specific indexes for your Spark jobs that cover exactly the data needed.
Ensure index replicas to avoid bottlenecks
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
Along with using covering indexes, make sure your indexes are replicated across multiple index nodes. Index replication not only provides high availability, but also allows queries to be load-balanced across index copies on different nodes for higher throughput. In practice, if you have (for example) 3 index nodes, replicating important indexes across them means the Spark connector’s parallel queries can hit different index nodes rather than all pounding a single node.
Tuning your write operations
Prefer the Data service for bulk writes
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
We recommend to use the key-value data source (Data service) rather than the Query service for write operations. Writing through the Data service (direct KV upserts) is typically several times faster than doing SQL++-based inserts. In fact, internal benchmarks have shown writing via KV can be around 3x faster than using SQL++ in Spark jobs. This is because the Data service can ingest documents in parallel directly to the nodes responsible, with lower latency per operation. Note that you have indices updated separately, if needed, for those new documents, as KV writes won’t automatically trigger index updates beyond the primary index.
Increase write partitions for Query service writes
(Couchbase Capella (DBaaS), base de datos operacional autogestionada)
While not recommended, if you decide to use couchbase.query for writing (for example, if performing a server side transformations while writing) , optimize the performance by using a high number of write partitions. You can repartition your DataFrame before writing so that Spark runs many concurrent write tasks. A rough guideline is to use on the order of hundreds of partitions for large scale writes via SQL++. For instance, using about 128 partitions per Query node CPU is a starting point some users have found effective. This means if you have 8 query cores, try ~1024 partitions. The idea is to flood the query service with enough parallel UPSERT statements to maximize throughput. Be cautious and find the right balance for your cluster – too high concurrency could overload the query service. Monitor Couchbase’s query throughput and adjust accordingly.
By following these tuning tips – aligning partition counts with cluster resources, indexing smartly, and choosing the right service for the job – you can achieve optimal performance for Couchbase-Spark integration. Keep an eye on both Spark’s job metrics and Couchbase’s performance stats (available in the Couchbase UI and logs) to identify any bottlenecks (e.g., if one query node is doing all the work, or if the network is saturated) and adjust the configuration as needed.
Community and support
Couchbase PySpark support is built upon Couchbase Spark Connector for Couchbase and is open-source, and we encourage you to contribute, provide feedback, and join the conversation. you can access our comprehensive documentación, join the Foros de Couchbase o Couchbase Discord.
Further reading
For more information and detailed documentation, please refer to the official Couchbase Spark Connector documentation and relevant section about PySpark:
- Couchbase PySpark Documentation
- Couchbase Spark Connector GitHub Repository
- Couchbase Forums (Spark Connector section)
- Couchbase PySpark Connector Jupyter Notebook Example
- Couchbase PySpark ML Example Jupyter Notebook: Hotel Cancellations
Happy coding!
The Couchbase Team
Deja un comentario
Lo siento, debes estar conectado para publicar un comentario.