run_polars.ipynb¶

Source revision: 75f65139e26eb7079b7a897c877de97859c0120b

This page is the generated source; it shows the complete readable projection of the canonical file.

This page is a generated, non-executed projection of the notebook.

The raw .ipynb file is the canonical notebook source; execution counts and outputs are intentionally omitted.

DataCoolie Fabric native Python/Polars runner¶

Run this notebook in a Fabric Python environment with the verified DataCoolie dependencies attached. Paths must use the native Fabric/OneLake namespace supported by the selected runtime.

METADATA_PATH = "/lakehouse/default/Files/datacoolie-example/metadata/metadata.json"
CONNECTIONS_PATH = None
SCHEMA_HINTS_PATH = None
WATERMARK_BASE_PATH = "/lakehouse/default/Files/datacoolie-example/.runtime/watermarks"
LOG_BASE_PATH = "/lakehouse/default/Files/datacoolie-example/.runtime/logs"
STAGE = ""
JOB_NUM = 1
JOB_INDEX = 0
RUN_ATTRIBUTES = {}

from datacoolie.core.models.run_config import DataCoolieRunConfig
from datacoolie.engines.polars_engine import PolarsEngine
from datacoolie.metadata.file_provider import FileProvider
from datacoolie.orchestration.driver import DataCoolieDriver
from datacoolie.platforms.fabric_platform import FabricPlatform

platform = FabricPlatform(runtime="fabric")
engine = PolarsEngine(platform=platform)
metadata = FileProvider(
    config_path=METADATA_PATH,
    connections_path=CONNECTIONS_PATH,
    schema_hints_path=SCHEMA_HINTS_PATH,
    platform=platform,
    watermark_base_path=WATERMARK_BASE_PATH,
)
config = DataCoolieRunConfig(
    job_num=int(JOB_NUM),
    job_index=int(JOB_INDEX),
    run_attributes=RUN_ATTRIBUTES,
    stop_on_error=True,
    allowed_function_prefixes=[],
)

with DataCoolieDriver(
    engine=engine,
    platform=platform,
    metadata_provider=metadata,
    log_base_path=LOG_BASE_PATH,
    config=config,
) as driver:
    result = driver.run(stage=STAGE or None)
    if result.failed:
        raise RuntimeError(f"DataCoolie run failed for {result.failed} dataflows")

print({"job_id": config.job_id, "total": result.total, "failed": result.failed})