DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
SQL Operator Context (via DbApiHook integration):
| Code Block | ||||
|---|---|---|---|---|
| ||||
# Automatically injected context for PostgreSQL:
{
"database_type": "postgresql",
"version": "15.2",
"available_tables": ["customers", "orders", "products"],
"schema_info": {
"customers": {
"customer_id": {"type": "integer", "nullable": False, "primary_key": True},
"email_address": {"type": "varchar(255)", "nullable": False, "unique": True},
"total_revenue": {"type": "decimal(10,2)", "nullable": True}
}
},
"sample_data": {
"customers": [
{"customer_id": 1, "email_address": "john@example.com", "total_revenue": 1250.00},
{"customer_id": 2, "email_address": "jane@example.com", "total_revenue": 890.50}
]
},
"dialect_features": {
"supports_window_functions": True,
"supports_cte": True,
"date_functions": ["DATE_TRUNC", "EXTRACT", "AGE"]
}
}
File Operator Context (via S3Hook/GCSHook integration):
# Automatically injected context for S3 Parquet files:
{
"storage_type": "s3",
"file_format": "parquet",
"file_size_mb": 245,
"estimated_rows": 1000000,
"schema_info": {
"id": "int64",
"name": "string",
"email": "string",
"signup_date": "timestamp[ns]"
},
"sample_data": [
{"id": 1, "name": "John Doe", "email": "john@example.com"},
{"id": 2, "name": "Jane Smith", "email": "jane@example.com"}
],
"partitioning": ["year", "month"],
"compression": "snappy"
} |
...
SQL Operator Built-in Protection:
| Code Block | ||||
|---|---|---|---|---|
| ||||
class LLMSQLQueryOperator(BaseOperator):
# Built-in dangerous operation blocking
BLOCKED_KEYWORDS = ["DROP", "TRUNCATE", "DELETE FROM", "ALTER TABLE", "GRANT", "REVOKE"]
DEFAULT_SYSTEM_PROMPT = """You are a SQL expert integrated with {database_type}. |
...
File Operator Built-in Protection:
| Code Block | ||||
|---|---|---|---|---|
| ||||
class LLMFileAnalysisOperator(BaseOperator):
ALLOWED_OPERATIONS = ["read", "analyze", "summarize", "validate"]
DEFAULT_SYSTEM_PROMPT = """You are a file analysis expert for {storage_type} {file_format} files. |
...
Option A: Embedded HITL
| Code Block | ||||
|---|---|---|---|---|
| ||||
quality_check = LLMDataQualityOperator( task_id="customer_quality_analysis", data_sources=[customer_s3], prompt="Generate data quality validation queries", require_approval=True, # Built-in HITL approval_timeout=timedelta(hours=2) ) |
...
Option B: Separate HITL Steps
| Code Block | ||||
|---|---|---|---|---|
| ||||
# Generate queries
generate_queries = LLMDataQualityOperator(
task_id="generate_quality_queries",
data_sources=[customer_s3],
prompt="Generate data quality validation queries",
dry_run=True # Don't execute, just generate
)
# Human approval step
approve_queries = ApprovalOperator(
task_id="approve_queries",
body="{{ ti.xcom_pull(task_ids='generate_quality_queries') }}",
allow_modifications=True # Users can edit generated queries
)
# Execute approved queries
execute_analysis = AnalyticsOperator(
task_id="execute_quality_checks",
query="{{ ti.xcom_pull(task_ids='approve_queries') }}",
data_sources=[customer_s3]
)
generate_queries >> approve_queries >> execute_analysis |
...