{"id":17010,"date":"2025-04-07T10:48:50","date_gmt":"2025-04-07T17:48:50","guid":{"rendered":"https:\/\/www.couchbase.com\/blog\/?p=17010"},"modified":"2025-07-08T09:15:31","modified_gmt":"2025-07-08T16:15:31","slug":"pyspark-ga-couchbase-spark-connector","status":"publish","type":"post","link":"https:\/\/www.couchbase.com\/blog\/ko\/pyspark-ga-couchbase-spark-connector\/","title":{"rendered":"Couchbase\uc640 PySpark\ub85c \ud655\uc7a5\uc131\uc774 \ub6f0\uc5b4\ub09c AI\/ML \uc560\ud50c\ub9ac\ucf00\uc774\uc158 \uad6c\ucd95\ud558\uae30"},"content":{"rendered":"

We are excited to announce the General Availability (GA) of the Python support for Couchbase Spark Connector<\/a>, bringing first-class integration between Couchbase Server and Apache Spark<\/a> to Python data engineers\u200b. This GA release means the connector is production-ready and fully supported, enabling PySpark applications to seamlessly read from and write to Couchbase. With Couchbase\u2019s high-performance NoSQL database (with SQL++\/SQL++ query language) and Spark\u2019s distributed processing engine, data engineers can now easily combine these technologies to build fast, scalable data pipelines and analytics workflows. In short, the Couchbase Spark Connector for PySpark unlocks efficient, parallel data integration \u2013 allowing you to leverage Spark for ETL\/ELT, real-time analytics, machine learning, and more on data stored in Couchbase.<\/p>\n

In this post, we\u2019ll cover how to get started with the PySpark connector, demonstrate basic read\/write operations (both key-value and query-based) for both Couchbase operational database and Capella Columnar databases; and share performance tuning tips to get the best throughput. Whether you\u2019ve been using the Couchbase Spark Connector in Scala, or if you\u2019re new to Couchbase-Spark integration, this guide will help you quickly ramp up using PySpark for your data engineering needs.<\/p>\n

Why PySpark?<\/h2>\n

Adding PySpark support to the existing Couchbase Spark Connector was driven by the growing demand from data engineers and developers who prefer Python for its simplicity and massive Python ML ecosystem for Spark in data science and engineering workflows. This support ensures that teams already using Python can now integrate Couchbase (whether you are using Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a> or Capella Columnar<\/a> database) into Python-based Spark workflows, enabling broader adoption and streamlined data processes.<\/p>\n

Python\u2019s dominance in AI\/ML use cases, supported by frameworks such as SparkML, PyTorch, TensorFlow, H2O, DataRobot, scikit-learn, and SageMaker, as well as popular exploratory data analysis tools like Matplotlib and Plotly, further underscores the necessity for PySpark integration. Additionally, PySpark compatibility unlocks accelerated ETL and ML pipelines leveraging GPU acceleration (Spark RAPIDS) and facilitates sophisticated feature engineering and data wrangling tasks using widely adopted libraries such as Pandas, NumPy, and Spark’s built-in feature engineering APIs. This new support significantly streamlines data processes and expands adoption opportunities for Couchbase in data science and engineering teams.<\/p>\n

Getting started with Couchbase PySpark<\/h2>\n

Getting started is straightforward. The Couchbase Spark Connector is distributed as a single JAR (Java archive) that you add to your Spark environment. You can obtain the connector from the official Couchbase download site<\/a> or via Maven coordinates<\/a>. Once you have the JAR, using it in PySpark is as simple as configuring your Spark session with the connector and Couchbase connection settings.<\/p>\n

1. Get or create a Couchbase operational database or Capella Columnar database<\/b><\/p>\n

The fastest way to start with Couchbase is to use our Capella DBaaS<\/a>. Once there, you can either find your existing database or create an operational<\/a> or columnar<\/a> (for analytics) database. Alternatively, you can use our self managed Couchbase<\/a>.<\/p>\n

2. Install PySpark (if not already)<\/b><\/p>\n

If you are working in a Python environment, install PySpark using pip. For example, in a virtual environment:<\/p>\n

pip install pyspark<\/pre>\n
This will install Apache Spark for use with Python. If you\u2019re running on an existing Spark cluster or Databricks, PySpark may already be available.<\/p>\n
3. Include the Couchbase Spark Connector JAR<\/b><\/p>\n
Download<\/a> the spark-connector-assembly-<version>.jar<\/code> for the latest connector release. Then, when creating your Spark session or submitting your job, provide this JAR in the configuration. You can do this by setting the --jars<\/code> option in spark-submit<\/code> or via the SparkSession builder in code (as shown below).<\/p>\n
4. Configure the Couchbase connection<\/b><\/p>\n
You need to specify the Couchbase cluster connection string and credentials (username and password). In Capella, you can find this on the \u201cConnect\u201d tab for operational and Settings->Connection String<\/strong> for columnar. Optionally, specify a default bucket or scope if needed (though you can also specify bucket\/scope per operation).<\/p>\n
Below is a quick PySpark example<\/b> that sets up a SparkSession<\/code> to connect to a Couchbase cluster and then reads some data:<\/p>\n
from pyspark.sql import SparkSession\r\n# Initialize SparkSession with Couchbase connector and connection settings\r\n\r\nspark = SparkSession.builder \\\r\n\u00a0\u00a0\u00a0\u00a0.appName(\"CouchbaseIntegrationExample\") \\\r\n\u00a0\u00a0\u00a0\u00a0.master(\"local[*]\") \\\u00a0 # using local Spark for example; omit or adjust for Spark cluster\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.jars\", \"\/path\/to\/spark-connector-assembly-<version>.jar\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.connectionString\", \"couchbases:\/\/<YOUR_CLUSTER_HOSTNAME>\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.username\", \"<USERNAME>\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.password\", \"<PASSWORD>\") \\\r\n\u00a0\u00a0\u00a0\u00a0.getOrCreate()\r\n\r\n# Test the connection by reading a few documents from Couchbase (using a sample bucket)\r\ndf = spark.read.format(\"couchbase.query\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"bucket\", \"bucket_name\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"scope_name\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"collection_name\") \\\r\n\u00a0\u00a0\u00a0\u00a0.load()\r\n\r\ndf.printSchema()\r\ndf.show(5)<\/pre>\n
In the code above, we configure the Spark session to include the Couchbase connector JAR and point it to a Couchbase cluster. We then create a DataFrame df<\/code> by reading from the bucket_name<\/code> bucket (specifically the scope_name.collection_name<\/code> collection) via the Query service.<\/p>\n
For the rest of this document, we are assuming you have loaded our sample dataset travel-sample <\/a>which can be done for Couchbase Capella operational<\/a> or Columnar<\/a> very easily.<\/p>\n
Read\/write to Couchbase using PySpark<\/h2>\n
Once your Spark session is connected to Couchbase, you can perform both key-value operations<\/b> (for writes) and query operations<\/b> (using SQL++ for both read and writes) through DataFrames.<\/p>\n
Following table shows the format Sparks connector supports to read and write to Couchbase and columnar databases:<\/p>\n\n\n\n\n\n
<\/td>\nCouchbase\/Capella operational database<\/b><\/td>\nCapella Columnar database<\/b><\/td>\n<\/tr>\n
Read operations<\/td>\nread.format(\"couchbase.query\")<\/code><\/td>\nread.format(\"couchbase.columnar\")<\/code><\/td>\n<\/tr>\n
Write operations<\/td>\n(recommended to use Data Service)<\/p>\n
write.format(\"couchbase.kv\")<\/code><\/p>\n
write.format(\"couchbase.query\")<\/code><\/td>\n
write.format(\"couchbase.columnar\")<\/code><\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n
Reading from Couchbase with a Query DataFrame<\/h3>\n
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<\/code>, 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:<\/p>\n
# Read all documents from a Couchbase collection using the Query service\r\nairlines_df = spark.read.format(\"couchbase.query\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"bucket\", \"travel-sample\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"inventory\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"airline\") \\\r\n\u00a0\u00a0\u00a0\u00a0.load()\r\n\r\n# Example: filter the DataFrame using Spark (will push down to Couchbase where possible)\r\nusa_airlines_df = airlines_df.filter(\"country = 'United States'\")\r\nusa_airlines_df.show(5)<\/pre>\n
In this example, airlines_df<\/code> loads all documents from the travel-sample.inventory.airline<\/code> 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<\/a> filters to Couchbase so that unnecessary data isn\u2019t transferred (i.e. it will include the WHERE country = 'United States'<\/code> clause in the SQL++ query it runs, if possible). The result, usa_airlines_df<\/code>, can be used like any other DataFrame in Spark (for example, you could join it with other DataFrames, apply aggregations, etc.).<\/p>\n
Under the hood, the connector partitions<\/a> the query results into multiple tasks if configured (more on this in Performance Tuning<\/i> below), and uses Couchbase\u2019s 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\u200b. This allows parallel reads from Couchbase, leveraging the distributed nature of both Spark and Couchbase.<\/p>\n
Writing to Couchbase with Key-Value (KV) operations (recommended)<\/h3>\n
The connector also supports writing data to Couchbase, either via the Data service (KV)<\/b> or via the Query service (executing SQL++ INSERT\/UPSERT<\/code> commands for you). The recommended<\/b> way for most use cases is to use the Key-Value data source<\/b> (format(\"couchbase.kv\")<\/code>) for better performance\u200b<\/a>. In key-value mode, each Spark task will write documents directly to Couchbase data nodes.<\/p>\n
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<\/code> (or META_ID<\/code> in newer versions) in the DataFrame for the document ID. You can also specify a custom ID field via the IdFieldName<\/code> option.<\/p>\n
For example, suppose we have a Spark DataFrame new_airlines_df<\/code> that we want to write to Couchbase. It has a column airline_id<\/code> that should serve as the Couchbase document key, and the rest of the columns are the document content:<\/p>\n
# Assume new_airlines_df is a DataFrame we want to write to Couchbase\r\n# It contains an \"airline_id\" column to use as the document ID.\r\nnew_airlines_df.write.format(\"couchbase.kv\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"bucket\", \"mybucket\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"myscope\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"airlines\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"idFieldName\", \"airline_id\") \\\r\n\u00a0\u00a0\u00a0\u00a0.save()<\/pre>\n
Writing to Couchbase with Query (SQL++) operations<\/h3>\n
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\")<\/code> 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.<\/p>\n
df.write.format(\"couchbase.query\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"bucket\", \"mybucket\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"myscope\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"airlines\") \\\r\n\u00a0\u00a0\u00a0\u00a0.mode(\"overwrite\") \\\r\n\u00a0\u00a0\u00a0\u00a0.save()<\/pre>\n
In the next section, let us modify these basic read\/write cases for Couchbase\u2019s latest analytics product – Capella Columnar.<\/p>\n
PySpark support for Capella \u200b\u200bColumnar<\/h2>\n
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<\/p>\n
Reading Columnar-Formatted Data with PySpark<\/h3>\n
Reading data from a Couchbase Capella Columnar cluster in PySpark is similar to couchbase operational cluster except three changes:<\/p>\n
    \n
  1. Use the format(\"couchbase.columnar\")<\/code> to specify connection is for columnar service.<\/li>\n
  2. The connection string for columnar can be retrieved from Capella UI.<\/li>\n
  3. 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<\/li>\n<\/ol>\n
    Once Spark is configured, you can use the Spark DataFrame reader API to load data from the columnar service:<\/p>\n
    from pyspark.sql import SparkSession\r\n# Initialize SparkSession with Couchbase configs (assuming connector jar is available)\r\nspark = SparkSession.builder \\\r\n\u00a0\u00a0\u00a0\u00a0.appName(\"Couchbase Spark Connector Columnar Example\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.connectionString\", \"couchbases:\/\/your.columnar.connection.string\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.username\", \"YourColumnarUsername\") \\\r\n\u00a0\u00a0\u00a0\u00a0.config(\"spark.couchbase.password\", \"YourColumnarPassword\") \\\r\n\u00a0\u00a0\u00a0\u00a0.getOrCreate()\r\n\r\n# Read a DataFrame from Couchbase Capella Columnar (travel-sample.inventory.airline collection)\r\nairlines_df = spark.read.format(\"couchbase.columnar\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"database\", \"travel-sample\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"inventory\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"airline\") \\\r\n\u00a0\u00a0\u00a0\u00a0.load()<\/pre>\n
    In this example, the resulting airlines_df<\/code> is a normal Spark DataFrame \u2014 you can inspect it, run transformations, and perform actions like .count()<\/code> or .show()<\/code> as usual. For instance, airlines_df.show(5)<\/code> will print a few airline documents, and airlines_df.count()<\/code> 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)\u200b. All fields that consistently appear in the sampled documents become columns in the DataFrame, with appropriate Spark data types.<\/p>\n
    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)\u200b. 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.<\/p>\n
    Querying a Columnar Dataset in Couchbase via Spark<\/h3>\n
    Often you may not want to load an entire collection, especially if it\u2019s 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\", \"\")<\/code> to apply a SQL++ WHERE clause during the read operation. For instance, to load only airlines based in the United States:<\/p>\n
    usa_airlines_df = spark.read.format(\"couchbase.columnar\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"database\", \"travel-sample\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"scope\", \"inventory\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"collection\", \"airline\") \\\r\n\u00a0\u00a0\u00a0\u00a0.option(\"filter\", \"country = 'United States'\") \\\r\n\u00a0\u00a0\u00a0\u00a0.load()\r\n\r\nprint(usa_airlines_df.count())\u00a0 # Only airlines where country = 'United States'<\/pre>\n
    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 \u2013 the connector will offload simple aggregates<\/a> like COUNT<\/code>, MIN<\/code>, MAX<\/code>, and SUM<\/code> to the Columnar engine whenever possible, rather than computing them in Spark, for better performance\u200b<\/p>\n
    Once data is loaded into a DataFrame, you can perform standard Spark transformations<\/a>, 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:<\/p>\n
    airlines_df.createOrReplaceTempView(\"airlines_view\")\r\nresult_df = spark.sql(\"\"\"\r\n\u00a0\u00a0\u00a0\u00a0SELECT country, COUNT(*) AS airline_count\r\n\u00a0\u00a0\u00a0\u00a0FROM airlines_view\r\n\u00a0\u00a0\u00a0\u00a0GROUP BY country\r\n\u00a0\u00a0\u00a0\u00a0ORDER BY airline_count DESC\r\n\"\"\")\r\nresult_df.show(10)<\/pre>\n
    This query runs entirely within Spark engine, giving flexibility to integrate Couchbase data seamlessly into complex analytical workflows.<\/p>\n
    Having covered basic reads and writes, let\u2019s move on to how you can tune performance when moving large volumes of data between Couchbase and Spark.<\/p>\n
    Performance tuning tips<\/h2>\n
    To maximize throughput and efficiency when using the Couchbase PySpark Connector, consider the following best practices.<\/p>\n
    Tuning your read operations<\/h3>\n
    Use Query Partitioning for parallelism<\/b>
    \n(    Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a> or Capella Columnar<\/a>)<\/p>\n
    When reading via the Query service for operational or columnar database, take advantage of the connector\u2019s ability to partition the query results. You can specify a partitionCount<\/a> (and a numeric partitioning field with lower\/upper bounds) for the DataFrame read. A good rule of thumb is to set partitionCount<\/code> to at least the total number of query service CPU cores<\/b> available in your Couchbase cluster. This ensures Spark will run multiple queries in parallel, leveraging all query nodes. For example, if your Couchbase cluster\u2019s Query service has 8 cores in total, set partitionCount >= 8<\/code> 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.<\/p>\n
    Leverage covering indexes for query efficiency
    \n<\/b>(    Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a>)
    \n<\/b><\/p>\n
    If using SQL++ queries, try to query through covering indexes<\/a> whenever possible. A covering index is an index that includes all<\/i> fields your query needs, so the query can be served entirely from the index without fetching from the data service\u200b. Covered queries avoid the extra network hop to fetch full documents, thus delivering better performance<\/b>\u200b. Design your Couchbase secondary indexes to include the fields you filter on and<\/i> the fields you return, if feasible. This might mean creating specific indexes for your Spark jobs that cover exactly the data needed.<\/p>\n
    Ensure index replicas to avoid bottlenecks
    \n<\/b>(    Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a>)<\/p>\n
    Along with using covering indexes, make sure your indexes are replicated across multiple index nodes. Index replication<\/a> not only provides high availability, but also allows queries to be load-balanced across index copies on different nodes for higher throughput<\/b>\u200b. In practice, if you have (for example) 3 index nodes, replicating important indexes across them means the Spark connector\u2019s parallel queries can hit different index nodes rather than all pounding a single node.<\/p>\n
    Tuning your write operations<\/h3>\n
    Prefer the Data service for bulk writes<\/b>
    \n(    Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a>)<\/p>\n
    We recommend to use the key-value data source (Data service<\/a>) rather than the Query service for write operations. Writing through the Data service (direct KV upserts) is typically several times faster<\/b> than doing SQL++-based inserts. In fact, internal benchmarks have shown writing via KV can be around 3x faster<\/i> 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\u2019t automatically trigger index updates beyond the primary index.<\/p>\n
    Increase write partitions for Query service writes<\/b>
    \n(    Couchbase Capella (DBaaS)<\/a>, self managed operational database<\/a>)<\/p>\n
    While not recommended, if you decide to use couchbase.query<\/code> 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<\/b> for large scale writes via SQL++. For instance, using about 128 partitions per Query node CPU<\/i> 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 \u2013 too high concurrency could overload the query service. Monitor Couchbase\u2019s query throughput and adjust accordingly.<\/p>\n
    By following these tuning tips \u2013 aligning partition counts with cluster resources, indexing smartly, and choosing the right service for the job \u2013 you can achieve optimal performance for Couchbase-Spark integration. Keep an eye on both Spark\u2019s job metrics and Couchbase\u2019s 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.<\/p>\n
    Community and support<\/h2>\n
    Couchbase PySpark support is built upon Couchbase Spark Connector for Couchbase and is open-source<\/a>, and we encourage you to contribute, provide feedback, and join the conversation. you can access our comprehensive documentation<\/a>, join the Couchbase Forums<\/a> or Couchbase Discord<\/a>.<\/p>\n
    Further reading<\/h2>\n
    For more information and detailed documentation, please refer to the official Couchbase Spark Connector documentation<\/a> and relevant section about PySpark<\/a>:<\/p>\n