Structured Outputs in SQL with Columnar Typing 📊
A number of Databricks models support another responseFormat variant that allows you to supply a PySpark SQL Type as a formatting directive. This approach represents a compelling alternative to tool calling methods, offering direct schema integration within SQL workflows while maintaining the structured output capabilities your data pipelines require. 🚀
Tested Models that Support PySpark Type Response Formats ✅
databricks-meta-llama-3-3-70b-instruct🦙databricks-llama-4-maverick🦙databricks-claude-3-7-sonnet🤖databricks-claude-sonnet-4🤖databricks-gemma-3-12b💎
You can supply a STRING representation of the type or use the SCHEMA_OF_JSON function that accepts a JSON string representation of the desired response format. This flexibility enables both hardcoded schema definitions and dynamic schema generation based on your data requirements. 🔧
⚠️ Structural Quirk
One oddity of this approach is that the responseFormat must be defined as a STRUCT with a top-level key that contains the actual desired response format. For example: If you want {"response": "my content goes here"}, you'd need to wrap that into another JSON object like so: {"response_format": {"response": "my content goes here"}}. This 'top-level' JSON object must only have one key AND the name of the key must not be empty or a character that SQL would have issues processing. At the time of writing, this top-level key (response_format in my example) is discarded.
SELECT FROM_JSON(AI_QUERY(
endpoint => 'databricks-claude-sonnet-4',
request => "Generate a product review for a wireless Bluetooth headphone. Include the product name, overall rating out of 5 stars, and a concise summary review:
I recently purchased the Sony WH-1000XM4 Wireless Noise Canceling Headphones and I have to say I'm thoroughly impressed! The sound quality is absolutely phenomenal with deep, rich bass and crystal-clear highs. The active noise cancellation works like magic - I can barely hear anything from the outside world when I'm listening to music or on calls. The battery life is outstanding, lasting well over 24 hours on a single charge. The touch controls on the right ear cup are intuitive and responsive. My only minor complaint is that they can feel a bit heavy during extended wear sessions, but the comfort level is still quite good overall. The quick charge feature is a lifesaver when I forget to charge them overnight. I would definitely give these headphones a solid 4 out of 5 stars and highly recommend them to anyone looking for premium wireless headphones with excellent noise cancellation.",
responseFormat => SCHEMA_OF_JSON('{"ai_response": {"product_name":"Unknown Product","rating":0,"review_comment":"No review provided"}}')
), SCHEMA_OF_JSON('{"product_name":"Unknown Product","rating":0,"review_comment":"No review provided"}'))
Limitations 🚧
While often convenient, this method does introduce some complexity that becomes more apparent as your use cases grow in sophistication.
Lack of Context in the Response Format 📝
Unlike json_schema based approaches where you can provide field-level descriptions, ranges, and conditions, the SQL type approach requires you to rely on prompting instructions—explicitly embedding formatting requirements within your prompt text. This limitation becomes increasingly challenging as your schema complexity grows, since all validation logic must be communicated through natural language rather than structured schema definitions. 💬
We'll observe the results of this constraint as we increase the complexity of examples throughout this guide. 📈
Necessary Tools and Knowledge 🛠️
In addition to a basic understanding of AI_QUERY, there are a number of key functions that allow us to put the data on rails and ensure reliable schema handling:
- TO_JSON: Returns a JSON string with the STRUCT or VARIANT. 📤
- FROM_JSON: Returns a struct value with the jsonStr and schema. 📥
- SCHEMA_OF_JSON: Returns the schema of a JSON string in DDL format. 🔍
We'll compose these functions to help get our data in a strongly-typed format that's easier and more reliable to ETL in SQL workflows. This approach provides the schema validation and type safety that downstream data processing requires while maintaining the flexibility of natural language interaction. ⚡
🎯 When to Use Strong Typing
You don't always want to have strongly typed data in SQL - but most use cases will benefit from this approach. One such exception would be when working with responses of variadic shape, where you're expecting different keys from response to response. In this case, it's often beneficial to leave the data in STRING textual format or use the VARIANT data type that provides flexible schema evolution capabilities. 🔄
Python + PySpark SQL 🐍
The remainder of this guide will be based on examples (simple, intermediate, complex) that explore how using SQL Types for Response Format behave, where they shine, and their limits. Keep in mind the prompts are simple and unoptimized—production implementations should include more sophisticated prompt engineering techniques. 🚀
By using Python with PySpark SQL (the reader is completely welcome to implement this directly in SQL), we can leverage Pydantic for conversions to JSON Schema and generating response previews while still getting all the benefits of PySpark SQL to make formatting the prompts less painful. This hybrid approach combines the flexibility of Python tooling with the performance and integration benefits of SQL-native processing. 🔀
For future use, you can use some of the patterns we've covered previously (like wrapping up most of this into a temporary function) to make your code more modular and readable for production deployments. 📦
Getting Started 🏁
We'll create a PreviewMixin class that makes it easier to generate an empty "preview" of the expected response shape without any meaningful data. This preview capability becomes essential for schema generation and testing throughout the development process. 👀
class PreviewMixin:
"""
A mixin class that provides a 'preview' class method for Pydantic models.
"""
@classmethod
def preview(self) -> dict:
return self.model_validate({}).model_dump_json()
A Simple Example 🎯
We'll embed the prompt and the query into the top-level model structure. This design pattern makes it easier to compose the final query used by PySpark while maintaining clear separation between data models and business logic. The encapsulation also improves maintainability by keeping related prompt logic co-located with the corresponding response schema. 🏗️
# Introductory Example: Simple Product Review 🎧
class ProductReview(PreviewMixin, BaseModel):
product_name: str = "Unknown Product"
rating: int = 0
review_comment: str = "No review provided"
@staticmethod
def prompt():
return 'Generate a product review for a wireless Bluetooth headphone. Include the product name, overall rating out of 5 stars, and a concise summary review:'
@staticmethod
def query():
return "I recently purchased the Sony WH-1000XM4 Wireless Noise Canceling Headphones and I have to say I'm thoroughly impressed! The sound quality is absolutely phenomenal with deep, rich bass and crystal-clear highs. The active noise cancellation works like magic - I can barely hear anything from the outside world when I'm listening to music or on calls. The battery life is outstanding, lasting well over 24 hours on a single charge. The touch controls on the right ear cup are intuitive and responsive. My only minor complaint is that they can feel a bit heavy during extended wear sessions, but the comfort level is still quite good overall. The quick charge feature is a lifesaver when I forget to charge them overnight. I would definitely give these headphones a solid 4 out of 5 stars and highly recommend them to anyone looking for premium wireless headphones with excellent noise cancellation."
Usage in SQL:
model = 'databricks-claude-3-7-sonnet'
preview = ProductReview.preview()
response_format = f'{{"ai_response": {preview}}}'
prompt = ProductReview.prompt() + ':\n\n' + ProductReview.query()
query = f'''
SELECT FROM_JSON(AI_QUERY(
endpoint => '{model}',
request => "{prompt}",
responseFormat => SCHEMA_OF_JSON('{response_format}')
), SCHEMA_OF_JSON('{preview}'))
'''
display(
ai_response := spark.sql(query)
)
An Intermediate Example 🍳
As we move to more complex data structures, we begin to see both the strengths and limitations of the SQL type approach. This intermediate example demonstrates nested object handling and array processing—capabilities that showcase the method's flexibility while revealing areas where explicit schema definitions become more critical. 📊
# Intermediate Example: Recipe with Ingredients 🥗
class Ingredient(PreviewMixin, BaseModel):
name: str = Field(default="Unknown ingredient", description="Name of the ingredient")
quantity: str = Field(default="0", description="Amount needed (e.g., '2 cups', '1 tbsp')")
@classmethod
def preview(self) -> dict:
return self.model_validate({}).model_dump_json()
class Recipe(PreviewMixin, BaseModel):
name: str = Field(default="Untitled Recipe", description="Name of the recipe")
cooking_time_minutes: int = Field(default=0, description="Total cooking time in minutes")
difficulty_level: str = Field(default="beginner", description="Cooking difficulty: beginner, intermediate, or advanced")
ingredients: List[Ingredient] = Field(default_factory=lambda: [Ingredient()], description="List of required ingredients with quantities")
instructions: List[str] = Field(default_factory=lambda: [''], description="Step-by-step cooking instructions")
@staticmethod
def prompt():
return 'Create a recipe for a healthy breakfast dish. Include the recipe name, cooking time in minutes, difficulty level (beginner/intermediate/advanced), a list of ingredients with quantities, and step-by-step cooking instructions.'
@staticmethod
def query():
return """
Mediterranean Quinoa Breakfast Bowl 🌅
This nutritious breakfast bowl combines protein-rich quinoa with fresh Mediterranean flavors for a satisfying start to your day. Perfect for meal prep and customizable to your taste preferences. 🥙
**Cooking Time:** 25 minutes ⏰
**Difficulty:** Intermediate 👨🍳
**Ingredients:** 🛒
- 1 cup quinoa, rinsed
- 2 cups vegetable broth
- 4 large eggs
- 1/4 cup extra virgin olive oil
- 2 medium tomatoes, diced
- 1 cucumber, chopped
- 1/2 red onion, finely sliced
- 1/2 cup kalamata olives, pitted and halved
- 4 oz feta cheese, crumbled
- 1/4 cup fresh parsley, chopped
- 2 tablespoons fresh lemon juice
- 1 teaspoon dried oregano
- 1/2 teaspoon salt
- 1/4 teaspoon black pepper
- 2 tablespoons pine nuts, toasted
**Instructions:** 📋
1. Rinse quinoa thoroughly under cold water until water runs clear. In a medium saucepan, bring vegetable broth to a boil. 💧
2. Add quinoa to boiling broth, reduce heat to low, cover and simmer for 15 minutes until liquid is absorbed. Remove from heat and let stand 5 minutes, then fluff with a fork. 🍚
3. While quinoa cooks, bring a small pot of water to boil. Carefully add eggs and cook for 7 minutes for soft-boiled eggs. Transfer to ice water bath to stop cooking. 🥚
4. In a large bowl, whisk together olive oil, lemon juice, oregano, salt, and pepper to make dressing. 🥗
5. Add warm quinoa to the dressing and toss to combine. Let cool slightly. 🌡️
6. Add diced tomatoes, cucumber, red onion, and olives to the quinoa mixture. Gently fold in crumbled feta and fresh parsley. 🍅
7. Divide mixture between 4 bowls. Peel soft-boiled eggs, cut in half, and place on top of each bowl. 🍽️
8. Sprinkle with toasted pine nuts and serve immediately, or refrigerate for up to 2 days for meal prep. 🌰
This Mediterranean quinoa breakfast bowl provides complete protein, healthy fats, and plenty of vegetables to fuel your morning! ⚡
"""
Usage in SQL:
model = 'databricks-llama-4-maverick'
preview = Recipe.preview()
response_format = f'{{"ai_response": {preview}}}'
prompt = Recipe.prompt() + ':\n\n' + Recipe.query()
query = f'''
SELECT FROM_JSON(AI_QUERY(
endpoint => '{model}',
request => "{prompt}",
responseFormat => SCHEMA_OF_JSON('{response_format}')
), SCHEMA_OF_JSON('{preview}'))
'''
display(
ai_response := spark.sql(query)
)
A More Complex Example 🏗️
This complex example pushes the boundaries of what's practical with SQL type response formatting. As data structures become deeply nested with multiple relationships and complex validation requirements, we begin to see where the limitations of prompt-only schema communication become most apparent. This example serves as both a demonstration of capability and a cautionary tale about complexity management. ⚠️
# This might be TOO complex for reliable results with prompt-only schema communication ⚡
# Complex Example: Software Project Analysis 💻
class TeamMember(PreviewMixin, BaseModel):
name: str = Field(default="Anonymous", description="Team member's name")
role: str = Field(default="Developer", description="Role in the project (e.g., Lead Developer, DevOps Engineer)")
experience_years: int = Field(default=0, description="Years of relevant experience")
skills: List[str] = Field(default_factory=lambda: [''], description="List of technical skills")
class ServiceDependency(PreviewMixin, BaseModel):
service_name: str = Field(default="unknown-service", description="Name of the dependent service")
dependency_type: str = Field(default="synchronous", description="Type of dependency (synchronous, asynchronous, database)")
criticality: str = Field(default="low", description="Dependency criticality: low, medium, high")
class MicroService(PreviewMixin, BaseModel):
name: str = Field(default="unnamed-service", description="Service name")
language: str = Field(default="unknown", description="Programming language used")
port: int = Field(default=8080, description="Service port number")
dependencies: List[ServiceDependency] = Field(default_factory=lambda: [ServiceDependency()], description="List of service dependencies")
resource_requirements: Dict[str, str] = Field(
default_factory=lambda: {"cpu": "0.5", "memory": "512Mi"},
description="Resource requirements (CPU, memory, etc.)"
)
class DeploymentConfig(PreviewMixin, BaseModel):
cloud_provider: str = Field(default="gcp", description="Cloud provider (gcp, aws, azure)")
kubernetes_version: str = Field(default="1.28", description="Kubernetes cluster version")
environment: str = Field(default="development", description="Deployment environment")
scaling_policy: str = Field(default="manual", description="Scaling policy: manual, horizontal, vertical")
backup_strategy: str = Field(default="none", description="Data backup and recovery strategy")
class RiskAssessment(PreviewMixin, BaseModel):
risk_category: str = Field(default="technical", description="Category of risk (technical, operational, security)")
severity: str = Field(default="low", description="Risk severity: low, medium, high, critical")
probability: str = Field(default="low", description="Likelihood of occurrence: low, medium, high")
description: str = Field(default="No risk identified", description="Detailed description of the risk")
mitigation_strategy: str = Field(default="Monitor regularly", description="Strategy to mitigate or manage the risk")
class ProjectAnalysis(PreviewMixin, BaseModel):
project_name: str = Field(default="Unnamed Project", description="Name of the software project")
description: str = Field(default="No description provided", description="Brief project description")
start_date: Optional[datetime] = Field(default=datetime.now(), description="Project start date")
estimated_completion: Optional[datetime] = Field(default=datetime.now(), description="Estimated completion date")
team: List[TeamMember] = Field(default_factory=lambda: [TeamMember()], description="Project team members")
services: List[MicroService] = Field(default_factory=lambda: [MicroService()], description="Microservices in the architecture")
deployment: DeploymentConfig = Field(default_factory=DeploymentConfig, description="Deployment configuration")
risks: List[RiskAssessment] = Field(default_factory=lambda: [RiskAssessment()], description="Identified project risks and mitigation strategies")
overall_complexity_score: int = Field(default=0, description="Overall project complexity score (0-10)")
recommended_timeline_weeks: int = Field(default=0, description="Recommended project timeline in weeks")
@staticmethod
def prompt():
return 'Analyze a cloud-native microservices project. Provide project metadata, team information including roles and experience levels, technical architecture details with services and their dependencies, deployment configuration, and a comprehensive risk assessment with mitigation strategies.'
@staticmethod
def query():
return """
# E-Commerce Platform Modernization Project 🛒
## Project Overview 📊
Project Name: CloudCart Microservices Platform
Description: Modernization of legacy monolithic e-commerce platform into cloud-native microservices architecture on Google Cloud Platform. The project involves decomposing a 10-year-old PHP monolith serving 2M+ monthly active users into scalable, containerized microservices with event-driven architecture. 🔄
Timeline: Started January 15, 2024, estimated completion by September 30, 2024 📅
## Team Composition 👥
**Sarah Chen** - Lead Cloud Architect 🏗️
- 8 years experience in distributed systems
- Skills: GCP, Kubernetes, Istio service mesh, Terraform, Go, Python
**Marcus Rodriguez** - Senior DevOps Engineer 🚀
- 6 years DevOps experience
- Skills: CI/CD pipelines, GitOps, ArgoCD, Helm charts, monitoring with Prometheus/Grafana, Docker
**Priya Patel** - Senior Backend Developer 💻
- 7 years microservices development
- Skills: Java Spring Boot, Node.js, gRPC, Apache Kafka, PostgreSQL, Redis
**Alex Thompson** - Platform Security Engineer 🔒
- 5 years cloud security experience
- Skills: Zero-trust architecture, Workload Identity, Certificate management, security scanning, compliance
**Jennifer Wu** - Site Reliability Engineer 📈
- 4 years SRE experience
- Skills: Observability, incident response, chaos engineering, performance optimization
## Technical Architecture 🏛️
### Core Services: ⚙️
**User Service** 👤
- Language: Java Spring Boot
- Port: 8080
- Dependencies:
- PostgreSQL database (high criticality, synchronous)
- Redis cache (medium criticality, synchronous)
- Auth service (high criticality, synchronous)
- Resources: CPU 1.0 cores, Memory 1Gi
**Product Catalog Service** 📦
- Language: Node.js
- Port: 3000
- Dependencies:
- Elasticsearch (high criticality, synchronous)
- Image storage service (medium criticality, asynchronous)
- Inventory service (medium criticality, synchronous)
- Resources: CPU 0.5 cores, Memory 512Mi
**Order Processing Service** 📋
- Language: Go
- Port: 8090
- Dependencies:
- Payment service (high criticality, synchronous)
- Inventory service (high criticality, synchronous)
- Kafka message broker (high criticality, asynchronous)
- PostgreSQL (high criticality, synchronous)
- Resources: CPU 2.0 cores, Memory 2Gi
**Payment Service** 💳
- Language: Java Spring Boot
- Port: 8085
- Dependencies:
- External payment gateway (critical, synchronous)
- Fraud detection service (high criticality, synchronous)
- Audit logging service (medium criticality, asynchronous)
- Resources: CPU 1.5 cores, Memory 1.5Gi
**Notification Service** 📧
- Language: Python FastAPI
- Port: 8000
- Dependencies:
- Kafka consumer (high criticality, asynchronous)
- Email service provider (medium criticality, asynchronous)
- SMS gateway (low criticality, asynchronous)
- Resources: CPU 0.5 cores, Memory 512Mi
## Deployment Configuration ☁️
Cloud Provider: Google Cloud Platform 🌐
Kubernetes Version: 1.29 ⚓
Environment: Production deployment with staging and development environments 🚀
Scaling Policy: Horizontal Pod Autoscaling based on CPU/memory metrics and custom SLIs 📊
Backup Strategy: Automated daily backups to Cloud Storage with 30-day retention, point-in-time recovery for databases 💾
Infrastructure includes:
- GKE Autopilot clusters across 3 zones 🌍
- Cloud SQL for PostgreSQL with read replicas 🗄️
- Memorystore for Redis ⚡
- Cloud Pub/Sub for async messaging 📨
- Istio service mesh for traffic management 🕸️
- GitOps deployment with ArgoCD 🔄
## Risk Assessment ⚠️
**Service Mesh Complexity Risk** 🕸️
Category: Technical
Severity: High
Probability: Medium
Description: Istio service mesh configuration complexity could lead to networking issues, difficult debugging, and performance overhead impacting user experience
Mitigation: Implement comprehensive integration testing, establish clear troubleshooting runbooks, maintain service mesh expertise through training
**Database Migration Risk** 🗄️
Category: Operational
Severity: Critical
Probability: Medium
Description: Migrating from monolithic database to distributed data architecture poses risks of data inconsistency, transaction failures, and potential data loss during cutover
Mitigation: Implement dual-write pattern with eventual consistency checks, comprehensive backup strategy, rollback procedures, and gradual migration approach
**Dependency Cascade Failure** ⛓️
Category: Technical
Severity: High
Probability: Medium
Description: Synchronous service dependencies could cause cascade failures if payment or user services become unavailable, impacting entire platform
Mitigation: Implement circuit breakers, retry policies with exponential backoff, bulkhead pattern, and graceful degradation strategies
**Security Compliance Gap** 🔐
Category: Security
Severity: High
Probability: Low
Description: Microservices architecture increases attack surface with service-to-service communication, potentially exposing sensitive customer and payment data
Mitigation: Implement zero-trust networking with mTLS, Workload Identity, secret management with Secret Manager, regular security audits and penetration testing
**Team Knowledge Gap** 🧠
Category: Operational
Severity: Medium
Probability: Medium
Description: Limited team experience with cloud-native patterns and troubleshooting distributed systems could slow incident resolution and impact system reliability
Mitigation: Establish comprehensive monitoring and alerting, conduct chaos engineering exercises, provide ongoing training, and create detailed operational documentation
## Project Metrics 📏
Overall Complexity Score: 8/10 (High complexity due to distributed architecture, data migration, and multiple technology stacks) 🎯
Recommended Timeline: 36 weeks (includes development, testing, staged rollout, and team training) ⏱️
"""
Usage in SQL:
# model = 'databricks-claude-sonnet-4'
model = 'databricks-llama-4-maverick'
preview = ProjectAnalysis.preview()
response_format = f'{{"ai_response": {preview}}}'
prompt = ProjectAnalysis.prompt() + ':\n\n' + ProjectAnalysis.query()
query = f'''
SELECT FROM_JSON(AI_QUERY(
endpoint => '{model}',
request => "{prompt}",
responseFormat => SCHEMA_OF_JSON('{response_format}')
), SCHEMA_OF_JSON('{preview}'))
'''
display(
ai_response := spark.sql(query)
)
This complex example illustrates the practical limits of SQL type response formatting. While the approach works for simpler structures, deeply nested schemas with multiple object relationships require more sophisticated schema communication methods. For such use cases, tool calling with rich field descriptions often provides more reliable results and better validation capabilities. 🎯