原文
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import ColumnInfo, TableType
# The only line that differs from production
w = WorkspaceClient(host="http://localhost:8000", token="dev")
w.catalogs.create(name="sales")
w.schemas.create(name="store", catalog_name="sales")
w.tables.create(
name="orders", catalog_name="sales", schema_name="store",
table_type=TableType.MANAGED,
data_source_format="DELTA", storage_location="/data/sales/store/orders",
columns=[
ColumnInfo(name="id", type_text="BIGINT"),
ColumnInfo(name="customer", type_text="STRING"),
ColumnInfo(name="amount", type_text="DECIMAL(10,2)"),
],
)
wh = w.warehouses.create(name="dev", cluster_size="Small")
w.statement_execution.execute_statement(
warehouse_id=wh.id,
statement="INSERT INTO sales.store.orders VALUES (1, 'Ana', 459.90)",
)
result = w.statement_execution.execute_statement(
warehouse_id=wh.id,
statement="SELECT customer, amount FROM sales.store.orders",
)
print(result.result.data_array) # [['Ana', '459.90']] — really storedfrom pyspark.sql import SparkSession
UC = "http://localhost:8000"
spark = (
SparkSession.builder
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
# Unity Catalog connector — three-part names resolve against MiniLake
.config("spark.sql.catalog.spark_catalog", "io.unitycatalog.spark.UCSingleCatalog")
.config("spark.sql.catalog.spark_catalog.uri", UC)
.config("spark.sql.catalog.sales", "io.unitycatalog.spark.UCSingleCatalog")
.config("spark.sql.catalog.sales.uri", UC)
.getOrCreate()
)
# Real Delta files: parquet + _delta_log/, written by real Spark
spark.createDataFrame([(1, "a"), (2, "b")], ["id", "name"]) \
.write.format("delta").mode("overwrite").save("/data/delta/sales/store/events")
spark.sql("INSERT INTO sales.store.events VALUES (3, 'c')")
print(spark.table("sales.store.events").count()) # 3
# ...and the same bytes are readable through MiniLake's SQL API:
# SELECT e.name, o.customer
# FROM sales.store.events e JOIN sales.store.orders o ON o.id = e.idprovider "databricks" {
host = "http://localhost:8000"
token = "dev"
}
resource "databricks_catalog" "sandbox" {
name = "sandbox"
comment = "Local sandbox catalog"
}
resource "databricks_schema" "things" {
catalog_name = databricks_catalog.sandbox.name
name = "things"
}
resource "databricks_sql_endpoint" "compute" {
name = "local-compute"
cluster_size = "Small"
}
# terraform apply -> creates real catalogs and real DuckDB databases
# terraform destroy -> removes them
# Resources backed by APIs MiniLake does not emulate fail loudly with 501,
# instead of silently pretending to succeed.bundle:
name: my_pipeline
targets:
dev:
mode: development
workspace:
host: http://localhost:8000
resources:
jobs:
daily_load:
name: daily_load
tasks:
- task_key: main
spark_python_task:
python_file: ../src/load.py
# $ databricks bundle deploy -t dev
# $ databricks bundle run daily_load -t dev
#
# The CLI syncs files through the Workspace API, the provider creates the job,
# and `bundle run` executes it on real Spark in a sibling container —
# returning the job's actual output.# Is it up, and which services answered?
curl http://localhost:8000/_minilake/health
# Create a catalog through the Databricks API surface
curl -X POST http://localhost:8000/api/2.1/unity-catalog/catalogs \
-H 'Content-Type: application/json' \
-d '{"name": "sales"}'
# Run SQL and get real rows back
curl -X POST http://localhost:8000/api/2.0/sql/statements \
-H 'Content-Type: application/json' \
-d '{"warehouse_id": "dev", "statement": "SELECT 42 AS answer"}'
# Give the next test a clean workspace — no restart needed
curl -X POST http://localhost:8000/_minilake/reset