Synapse Spark and SQL Queries

Module: Integrate Azure Cosmos DB Solution

Section: Analytical Workloads

Lesson: Synapse Spark and SQL Queries

Introduction: Why Analytical Workloads Matter in Cosmos DB

When we talk about Azure Cosmos DB, most people immediately think of transactional processing—the high-speed, low-latency reads and writes that power applications like e-commerce carts, user profiles, or Internet of Things (IoT) device telemetry. However, data is rarely useful if it just sits in a transactional database. To make informed business decisions, you need to perform complex aggregations, trend analysis, and pattern recognition. This is where analytical workloads come into play.

In a traditional setup, you might export your Cosmos DB data to a data warehouse or a data lake for analysis. This process, often called ETL (Extract, Transform, Load), introduces latency, requires managing separate pipelines, and risks data staleness. Azure Synapse Link for Cosmos DB changes this paradigm by providing a cloud-native analytical store. This store allows you to run near-real-time analytics using Synapse Spark or Synapse SQL without impacting the performance of your transactional workloads.

Understanding how to bridge the gap between transactional data and analytical insights is a critical skill for any data engineer or architect. By mastering Synapse Spark and SQL queries against the Cosmos DB analytical store, you enable your organization to derive value from data the moment it is generated, rather than waiting for nightly batch jobs to finish. This lesson will guide you through the architecture, implementation, and best practices for running analytical workloads on Cosmos DB.

Understanding the Architecture: The Analytical Store

Before diving into the code, it is essential to understand how the data is stored. Cosmos DB uses a dual-storage model: the Transactional Store and the Analytical Store. The transactional store is row-based and optimized for point reads and writes. The analytical store, conversely, is a columnar-based storage format. Columnar storage is highly efficient for analytical queries because it allows the engine to read only the specific columns required for a calculation, rather than scanning entire rows.

When you enable Synapse Link on a Cosmos DB container, the system automatically synchronizes data from the transactional store to the analytical store. This synchronization happens near-real-time and is managed entirely by the platform, meaning you do not need to write custom code to keep the two stores in sync. This separation of concerns is the key to performance; your analytical queries hit the columnar store, leaving the row-based transactional store free to handle user-facing application traffic.

Callout: Transactional vs. Analytical Store The transactional store is optimized for high-concurrency, low-latency operations on individual records. It uses a row-based format ideal for CRUD (Create, Read, Update, Delete) operations. The analytical store is optimized for large-scale data processing and aggregations, utilizing a columnar format that minimizes I/O overhead for analytical scans. Because these stores are isolated, your complex analytical queries will never consume the Request Units (RUs) assigned to your transactional application.

Setting Up the Environment

To start working with analytical workloads, you must ensure your environment is configured correctly. This involves three primary steps: enabling Synapse Link on your Cosmos DB account, configuring your containers, and setting up the Azure Synapse workspace.

Step 1: Enable Synapse Link

1. Navigate to your Azure Cosmos DB account in the Azure portal.
2. In the left-hand menu, under the Integrations section, select Azure Synapse Link.
3. Click the Enable button. This action registers the feature for your account and prepares the infrastructure.

Step 2: Enable Analytical Store on the Container

1. Go to your Data Explorer in the Cosmos DB portal.
2. Select the container you wish to analyze.
3. Go to Settings and look for the Analytical Store setting.
4. Set it to On. You can choose to store all properties or select specific paths if you want to optimize storage costs.

Step 3: Link to Synapse Workspace

1. Open your Azure Synapse Workspace.
2. Navigate to the Manage hub and select Linked services.
3. Click New and select Azure Cosmos DB.
4. Provide the connection details for your account and verify the connection.

Once these steps are complete, your container's data will begin flowing into the analytical store. Note that historical data will be synchronized initially, and new data will follow automatically.

Querying with Synapse SQL (Serverless)

Synapse SQL serverless is an incredibly powerful tool for ad-hoc analysis. It allows you to query your analytical store using standard T-SQL syntax. You do not need to provision clusters or manage infrastructure; you simply write your SQL, and the service scales automatically to handle the query complexity.

Practical Example: Aggregating Sales Data

Imagine you have a container storing order documents. You want to calculate the total revenue per product category for the last month. Since the data is in the analytical store, you can query it directly using the OPENROWSET function.

SELECT 
    category, 
    SUM(price * quantity) AS total_revenue
FROM 
    OPENROWSET(
        'CosmosDB',
        'Account=my-cosmos-account;Database=RetailDB;Container=Orders',
        'SELECT * FROM c'
    ) AS [Orders]
GROUP BY category;

In this snippet, the OPENROWSET function acts as the bridge. It connects to the Cosmos DB container and presents the JSON data as a relational table. The SQL engine automatically infers the schema from the JSON properties, allowing you to treat your document-based data as if it were stored in a standard relational database.

Tip: Schema Inference When using OPENROWSET with Cosmos DB, the SQL engine performs schema inference based on the first few thousand rows. If your data is highly heterogeneous (i.e., different documents have vastly different structures), you may need to explicitly define the schema using a WITH clause to ensure data types are mapped correctly.

Querying with Synapse Spark

While SQL is excellent for structured reporting and quick ad-hoc analysis, Apache Spark is the gold standard for data engineering, complex transformations, and machine learning pipelines. Synapse Spark allows you to load your Cosmos DB analytical store data into a Spark DataFrame, where you can manipulate it using Python, Scala, or C#.

Practical Example: Calculating Moving Averages

Suppose you want to compute a 7-day moving average of stock prices stored in Cosmos DB. This is a classic time-series analysis task that is much easier to perform in Spark.

# Loading data from Cosmos DB into a Spark DataFrame
df = spark.read \
    .format("cosmos.olap") \
    .option("spark.synapse.linkedService", "MyCosmosDBLinkedService") \
    .option("spark.cosmos.container", "StockPrices") \
    .load()

# Performing the transformation using PySpark
from pyspark.sql import Window
import pyspark.sql.functions as F

windowSpec = Window.partitionBy("symbol").orderBy("timestamp").rowsBetween(-6, 0)

moving_avg_df = df.withColumn(
    "moving_avg", 
    F.avg("price").over(windowSpec)
)

# Displaying the result
moving_avg_df.select("symbol", "timestamp", "price", "moving_avg").show()

This code snippet demonstrates the power of the Spark connector. By using .format("cosmos.olap"), we explicitly tell Spark to pull data from the analytical store. Once the data is in a DataFrame, we can use the full breadth of the Spark SQL and MLlib libraries to perform sophisticated calculations that would be difficult or impossible to write in standard T-SQL.

Performance Best Practices

Running analytical queries against a production system requires careful planning. While the analytical store is isolated from the transactional store, poorly written queries can still consume significant compute resources within the Synapse environment, leading to longer wait times or higher costs.

1. Optimize Data Projection

Always select only the columns you need. If your documents contain 50 fields but you only need three for your report, use a projection in your SQL query or Spark transformation. This reduces the amount of data read from the columnar storage and significantly improves performance.

2. Leverage Partitioning

If your Cosmos DB container is partitioned, ensure your analytical queries align with the partition strategy when possible. While the analytical store is not bound by the same request unit constraints as the transactional store, filtering by partition key can still help the engine narrow down the data scan, especially in very large datasets.

3. Use Materialized Views

If you find yourself running the same complex aggregation repeatedly, consider using Synapse pipelines to materialize the results into a separate table or a parquet file in Azure Data Lake Storage. This "pre-computation" approach is a staple of data engineering best practices and ensures that end-users experience sub-second response times for common dashboard queries.

Warning: Data Type Mismatches Cosmos DB is schema-agnostic, meaning a single field could contain a string in one document and an integer in another. When loading this data into Spark or SQL, these mismatches can cause query failures. Always sanitize your data or use explicit schema definitions to handle these inconsistencies before running analytical workloads.

Comparison of Synapse SQL vs. Spark

Choosing between Synapse SQL and Spark depends largely on your use case and the skill set of your team. The following table provides a quick reference to help you decide which tool to use.

Use SQL when you need to answer a quick business question or when you are building a dashboard in Power BI. Use Spark when you need to clean, transform, or join data from multiple sources before presenting the final result.

Common Pitfalls and How to Avoid Them

Pitfall 1: Over-reliance on "Select Star"

It is tempting to run SELECT * when exploring a new dataset. In the context of a massive analytical store, this is a performance killer. It forces the engine to materialize every field, including large nested objects or arrays that you likely don't need.

• The Fix: Always explicitly name the columns you need in your SQL queries. In Spark, use the .select() method immediately after loading the data to discard unnecessary columns early in the pipeline.

Pitfall 2: Ignoring Time-to-Live (TTL)

The analytical store consumes storage, and you are billed for that storage. If you do not manage the lifecycle of your data, costs can balloon.

• The Fix: Configure the Analytical TTL on your container. This allows you to automatically purge data from the analytical store that is older than a certain threshold (e.g., 90 days), keeping your storage costs predictable.

Pitfall 3: Failing to Monitor Query Performance

It is easy to assume that because the analytical store is "separate," it is immune to performance issues. However, queries that perform full table scans on petabyte-scale data will still be slow.

• The Fix: Use the Synapse Studio monitoring tools to track your query execution times. If a query is consistently slow, analyze the query plan (in SQL) or the Spark UI (in Spark) to identify bottlenecks such as data shuffling or excessive disk spills.

Step-by-Step: Building an End-to-End Analytical Pipeline

To solidify your understanding, let's walk through the process of creating a simple pipeline that reads from Cosmos DB, transforms the data, and saves the output to a data lake for downstream use.

1. Define the Linked Service: Ensure your Synapse workspace is linked to the Cosmos DB account.
2. Create a Notebook: In Synapse Studio, create a new Spark Notebook.
3. Load the Data:

   # Read from the analytical store
   raw_data = spark.read \
       .format("cosmos.olap") \
       .option("spark.synapse.linkedService", "MyLinkedService") \
       .option("spark.cosmos.container", "Sales") \
       .load()
   

4. Transform the Data:

   # Filter for completed orders and add a tax field
   processed_data = raw_data.filter(raw_data.status == "Completed") \
       .withColumn("tax_amount", raw_data.price * 0.08)
   

5. Write to Data Lake:

   # Save as Parquet for cost-effective storage
   processed_data.write.mode("overwrite").parquet("abfss://[email protected]/sales_processed")
   

6. Schedule the Pipeline: Use a Synapse Pipeline to trigger this notebook on a recurring schedule (e.g., every morning at 6:00 AM).

This workflow represents the standard industry practice for handling analytical workloads. By using the analytical store as the source, you ensure that your ETL process is fast and doesn't interfere with the transactional applications.

Advanced Considerations: Handling Nested JSON

One of the unique challenges of working with Cosmos DB is its document-based nature. Documents often contain nested structures, such as lists of items within an order. When you query this data, you must know how to "flatten" or "explode" these structures to make them usable for analysis.

In Spark, you can use the explode function to transform an array of items into individual rows.

from pyspark.sql.functions import explode

# Assuming 'items' is an array column in your document
flattened_df = df.select("order_id", explode("items").alias("item"))

This operation turns one document with five items into five separate rows. This is essential for calculating metrics like "average number of items per order" or "most popular individual product." Mastering these transformation functions is what separates a novice data engineer from an expert.

Callout: The Power of Columnar Storage The efficiency of the analytical store comes from its ability to ignore data it doesn't need. When you query a specific column, the engine only reads the blocks of data associated with that column. If your JSON documents have 100 fields, but your query only references 2, the analytical store provides a massive performance boost over the transactional store, which would be forced to read the entire JSON blob for every record.

Security and Governance

When exposing Cosmos DB data to analytical tools, security must remain a priority. Azure Synapse provides several layers of protection:

• Role-Based Access Control (RBAC): Use Azure Active Directory (Azure AD) to manage who can access the Synapse workspace and the underlying data.
• Network Isolation: Use Managed Virtual Networks in Synapse to ensure that your data remains within the Azure backbone and is not exposed to the public internet.
• Encryption: Data in the analytical store is encrypted at rest by default using service-managed keys or customer-managed keys (CMK), providing an additional layer of security for sensitive information.

Always follow the principle of least privilege. If a user only needs to run SQL queries for reporting, do not grant them permissions to edit Spark notebooks or manage pipelines.

Troubleshooting Common Issues

"The Analytical Store is Empty"

If you have enabled the analytical store but see no data, check the following:

1. Synchronization Lag: It can take a few minutes for the initial synchronization to start after enabling the store.
2. Container Settings: Double-check that the analytical store setting is actually set to "On" for the specific container you are querying.
3. Data Volume: If you have just created the container, ensure you have actually written some data to it. The system only synchronizes data that is written after the feature is enabled (plus any existing data).

"Query Returns Inconsistent Results"

If your queries return different results than your application, it might be due to the nature of the synchronization. The analytical store is "near-real-time," not "strictly real-time." There is typically a delay of a few seconds to a minute between a write in the transactional store and its appearance in the analytical store. For most analytical use cases, this is perfectly acceptable, but it is important to communicate this lag to your stakeholders.

Key Takeaways

As we conclude this lesson, keep these core principles in mind when designing your analytical solutions with Azure Cosmos DB:

1. Decoupling is Key: By using the analytical store, you effectively isolate your analytical workloads from your transactional workloads, ensuring that complex queries never impact the performance of your user-facing applications.
2. Choose the Right Tool: Use Synapse SQL for quick, structured reporting and ad-hoc analysis. Use Synapse Spark for intensive data transformation, complex engineering, and machine learning pipelines.
3. Optimize for Columnar Storage: Always project only the columns you need. Avoid "select all" patterns, as they increase I/O overhead and can lead to slower query performance.
4. Manage Your Lifecycle: Use Analytical TTL to automatically manage the size of your analytical store. This keeps storage costs down and ensures that your analytical data remains relevant.
5. Handle JSON Carefully: Be prepared to flatten nested data structures using functions like explode in Spark or by defining clear schemas in your SQL queries to handle the document-based nature of Cosmos DB.
6. Security First: Always leverage Azure AD RBAC and network isolation to secure your analytical environment, ensuring that data is only accessible to authorized users and services.
7. Monitor Performance: Regularly review your query execution plans and monitor the Synapse workspace to identify and resolve bottlenecks before they impact your business operations.

By following these practices, you can build a robust, scalable, and cost-effective analytical pipeline that allows your organization to turn raw data into actionable insights with minimal friction. The combination of Cosmos DB's flexible storage and Synapse's powerful compute capabilities is a formidable toolset for any modern data architecture.