Deepchecks

zenml.integrations.deepchecks

Deepchecks integration for ZenML.

The Deepchecks integration provides a way to validate your data in your pipelines. It includes a way to detect data anomalies and define checks to ensure quality of data.

The integration includes custom materializers to store and visualize Deepchecks SuiteResults.

Attributes

DEEPCHECKS = 'deepchecks' module-attribute

DEEPCHECKS_DATA_VALIDATOR_FLAVOR = 'deepchecks' module-attribute

PANDAS = 'pandas' module-attribute

Classes

DeepchecksIntegration

Bases: Integration

Definition of Deepchecks integration for ZenML.

Functions

activate() -> None classmethod

Activate the Deepchecks integration.

Source code in src/zenml/integrations/deepchecks/__init__.py

@classmethod
def activate(cls) -> None:
    """Activate the Deepchecks integration."""
    from zenml.integrations.deepchecks import materializers  # noqa

flavors() -> List[Type[Flavor]] classmethod

Declare the stack component flavors for the Deepchecks integration.

Returns:

Type	Description
List[Type[Flavor]]	List of stack component flavors for this integration.

Source code in src/zenml/integrations/deepchecks/__init__.py

@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Declare the stack component flavors for the Deepchecks integration.

    Returns:
        List of stack component flavors for this integration.
    """
    from zenml.integrations.deepchecks.flavors import (
        DeepchecksDataValidatorFlavor,
    )

    return [DeepchecksDataValidatorFlavor]

get_requirements(target_os: Optional[str] = None, python_version: Optional[str] = None) -> List[str] classmethod

Method to get the requirements for the integration.

Parameters:

Name	Type	Description	Default
target_os	Optional[str]	The target operating system to get the requirements for.	None
python_version	Optional[str]	The Python version to use for the requirements.	None

Returns:

Type	Description
List[str]	A list of requirements.

Source code in src/zenml/integrations/deepchecks/__init__.py

@classmethod
def get_requirements(
    cls, target_os: Optional[str] = None, python_version: Optional[str] = None
) -> List[str]:
    """Method to get the requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.
        python_version: The Python version to use for the requirements.

    Returns:
        A list of requirements.
    """
    from zenml.integrations.pandas import PandasIntegration

    return cls.REQUIREMENTS + \
        PandasIntegration.get_requirements(target_os=target_os, python_version=python_version)

Flavor

Class for ZenML Flavors.

Attributes

config_class: Type[StackComponentConfig] abstractmethod property

Returns StackComponentConfig config class.

Returns:

Type	Description
Type[StackComponentConfig]	The config class.

config_schema: Dict[str, Any] property

The config schema for a flavor.

Returns:

Type	Description
Dict[str, Any]	The config schema.

docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[StackComponent] abstractmethod property

Implementation class for this flavor.

Returns:

Type	Description
Type[StackComponent]	The implementation class for this flavor.

logo_url: Optional[str] property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
Optional[str]	The flavor logo.

name: str abstractmethod property

The flavor name.

Returns:

Type	Description
str	The flavor name.

sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

service_connector_requirements: Optional[ServiceConnectorRequirements] property

Service connector resource requirements for service connectors.

Specifies resource requirements that are used to filter the available service connector types that are compatible with this flavor.

Returns:

Type	Description
Optional[ServiceConnectorRequirements]	Requirements for compatible service connectors, if a service
Optional[ServiceConnectorRequirements]	connector is required for this flavor.

type: StackComponentType abstractmethod property

The stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

Functions

from_model(flavor_model: FlavorResponse) -> Flavor classmethod

Loads a flavor from a model.

Parameters:

Name	Type	Description	Default
flavor_model	FlavorResponse	The model to load from.	required

Raises:

Type	Description
CustomFlavorImportError	If the custom flavor can't be imported.
ImportError	If the flavor can't be imported.

Returns:

Type	Description
Flavor	The loaded flavor.

Source code in src/zenml/stack/flavor.py

@classmethod
def from_model(cls, flavor_model: FlavorResponse) -> "Flavor":
    """Loads a flavor from a model.

    Args:
        flavor_model: The model to load from.

    Raises:
        CustomFlavorImportError: If the custom flavor can't be imported.
        ImportError: If the flavor can't be imported.

    Returns:
        The loaded flavor.
    """
    try:
        flavor = source_utils.load(flavor_model.source)()
    except (ModuleNotFoundError, ImportError, NotImplementedError) as err:
        if flavor_model.is_custom:
            flavor_module, _ = flavor_model.source.rsplit(".", maxsplit=1)
            expected_file_path = os.path.join(
                source_utils.get_source_root(),
                flavor_module.replace(".", os.path.sep),
            )
            raise CustomFlavorImportError(
                f"Couldn't import custom flavor {flavor_model.name}: "
                f"{err}. Make sure the custom flavor class "
                f"`{flavor_model.source}` is importable. If it is part of "
                "a library, make sure it is installed. If "
                "it is a local code file, make sure it exists at "
                f"`{expected_file_path}.py`."
            )
        else:
            raise ImportError(
                f"Couldn't import flavor {flavor_model.name}: {err}"
            )
    return cast(Flavor, flavor)

generate_default_docs_url() -> str

Generate the doc urls for all inbuilt and integration flavors.

Note that this method is not going to be useful for custom flavors, which do not have any docs in the main zenml docs.

Returns:

Type	Description
str	The complete url to the zenml documentation

Source code in src/zenml/stack/flavor.py

def generate_default_docs_url(self) -> str:
    """Generate the doc urls for all inbuilt and integration flavors.

    Note that this method is not going to be useful for custom flavors,
    which do not have any docs in the main zenml docs.

    Returns:
        The complete url to the zenml documentation
    """
    from zenml import __version__

    component_type = self.type.plural.replace("_", "-")
    name = self.name.replace("_", "-")

    try:
        is_latest = is_latest_zenml_version()
    except RuntimeError:
        # We assume in error cases that we are on the latest version
        is_latest = True

    if is_latest:
        base = "https://docs.zenml.io"
    else:
        base = f"https://zenml-io.gitbook.io/zenml-legacy-documentation/v/{__version__}"
    return f"{base}/stack-components/{component_type}/{name}"

generate_default_sdk_docs_url() -> str

Generate SDK docs url for a flavor.

Returns:

Type	Description
str	The complete url to the zenml SDK docs

Source code in src/zenml/stack/flavor.py

def generate_default_sdk_docs_url(self) -> str:
    """Generate SDK docs url for a flavor.

    Returns:
        The complete url to the zenml SDK docs
    """
    from zenml import __version__

    base = f"https://sdkdocs.zenml.io/{__version__}"

    component_type = self.type.plural

    if "zenml.integrations" in self.__module__:
        # Get integration name out of module path which will look something
        #  like this "zenml.integrations.<integration>....
        integration = self.__module__.split(
            "zenml.integrations.", maxsplit=1
        )[1].split(".")[0]

        return (
            f"{base}/integration_code_docs"
            f"/integrations-{integration}/#{self.__module__}"
        )

    else:
        return (
            f"{base}/core_code_docs/core-{component_type}/"
            f"#{self.__module__}"
        )

to_model(integration: Optional[str] = None, is_custom: bool = True) -> FlavorRequest

Converts a flavor to a model.

Parameters:

Name	Type	Description	Default
integration	Optional[str]	The integration to use for the model.	None
is_custom	bool	Whether the flavor is a custom flavor.	True

Returns:

Type	Description
FlavorRequest	The model.

Source code in src/zenml/stack/flavor.py

def to_model(
    self,
    integration: Optional[str] = None,
    is_custom: bool = True,
) -> FlavorRequest:
    """Converts a flavor to a model.

    Args:
        integration: The integration to use for the model.
        is_custom: Whether the flavor is a custom flavor.

    Returns:
        The model.
    """
    connector_requirements = self.service_connector_requirements
    connector_type = (
        connector_requirements.connector_type
        if connector_requirements
        else None
    )
    resource_type = (
        connector_requirements.resource_type
        if connector_requirements
        else None
    )
    resource_id_attr = (
        connector_requirements.resource_id_attr
        if connector_requirements
        else None
    )

    model = FlavorRequest(
        name=self.name,
        type=self.type,
        source=source_utils.resolve(self.__class__).import_path,
        config_schema=self.config_schema,
        connector_type=connector_type,
        connector_resource_type=resource_type,
        connector_resource_id_attr=resource_id_attr,
        integration=integration,
        logo_url=self.logo_url,
        docs_url=self.docs_url,
        sdk_docs_url=self.sdk_docs_url,
        is_custom=is_custom,
    )
    return model

Integration

Base class for integration in ZenML.

Functions

activate() -> None classmethod

Abstract method to activate the integration.

Source code in src/zenml/integrations/integration.py

@classmethod
def activate(cls) -> None:
    """Abstract method to activate the integration."""

check_installation() -> bool classmethod

Method to check whether the required packages are installed.

Returns:

Type	Description
bool	True if all required packages are installed, False otherwise.

Source code in src/zenml/integrations/integration.py

@classmethod
def check_installation(cls) -> bool:
    """Method to check whether the required packages are installed.

    Returns:
        True if all required packages are installed, False otherwise.
    """
    for r in cls.get_requirements():
        try:
            # First check if the base package is installed
            dist = pkg_resources.get_distribution(r)

            # Next, check if the dependencies (including extras) are
            # installed
            deps: List[Requirement] = []

            _, extras = parse_requirement(r)
            if extras:
                extra_list = extras[1:-1].split(",")
                for extra in extra_list:
                    try:
                        requirements = dist.requires(extras=[extra])  # type: ignore[arg-type]
                    except pkg_resources.UnknownExtra as e:
                        logger.debug(f"Unknown extra: {str(e)}")
                        return False
                    deps.extend(requirements)
            else:
                deps = dist.requires()

            for ri in deps:
                try:
                    # Remove the "extra == ..." part from the requirement string
                    cleaned_req = re.sub(
                        r"; extra == \"\w+\"", "", str(ri)
                    )
                    pkg_resources.get_distribution(cleaned_req)
                except pkg_resources.DistributionNotFound as e:
                    logger.debug(
                        f"Unable to find required dependency "
                        f"'{e.req}' for requirement '{r}' "
                        f"necessary for integration '{cls.NAME}'."
                    )
                    return False
                except pkg_resources.VersionConflict as e:
                    logger.debug(
                        f"Package version '{e.dist}' does not match "
                        f"version '{e.req}' required by '{r}' "
                        f"necessary for integration '{cls.NAME}'."
                    )
                    return False

        except pkg_resources.DistributionNotFound as e:
            logger.debug(
                f"Unable to find required package '{e.req}' for "
                f"integration {cls.NAME}."
            )
            return False
        except pkg_resources.VersionConflict as e:
            logger.debug(
                f"Package version '{e.dist}' does not match version "
                f"'{e.req}' necessary for integration {cls.NAME}."
            )
            return False

    logger.debug(
        f"Integration {cls.NAME} is installed correctly with "
        f"requirements {cls.get_requirements()}."
    )
    return True

flavors() -> List[Type[Flavor]] classmethod

Abstract method to declare new stack component flavors.

Returns:

Type	Description
List[Type[Flavor]]	A list of new stack component flavors.

Source code in src/zenml/integrations/integration.py

@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Abstract method to declare new stack component flavors.

    Returns:
        A list of new stack component flavors.
    """
    return []

get_requirements(target_os: Optional[str] = None, python_version: Optional[str] = None) -> List[str] classmethod

Method to get the requirements for the integration.

Parameters:

Name	Type	Description	Default
target_os	Optional[str]	The target operating system to get the requirements for.	None
python_version	Optional[str]	The Python version to use for the requirements.	None

Returns:

Type	Description
List[str]	A list of requirements.

Source code in src/zenml/integrations/integration.py

@classmethod
def get_requirements(
    cls,
    target_os: Optional[str] = None,
    python_version: Optional[str] = None,
) -> List[str]:
    """Method to get the requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.
        python_version: The Python version to use for the requirements.

    Returns:
        A list of requirements.
    """
    return cls.REQUIREMENTS

get_uninstall_requirements(target_os: Optional[str] = None) -> List[str] classmethod

Method to get the uninstall requirements for the integration.

Parameters:

Name	Type	Description	Default
target_os	Optional[str]	The target operating system to get the requirements for.	None

Returns:

Type	Description
List[str]	A list of requirements.

Source code in src/zenml/integrations/integration.py

@classmethod
def get_uninstall_requirements(
    cls, target_os: Optional[str] = None
) -> List[str]:
    """Method to get the uninstall requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.

    Returns:
        A list of requirements.
    """
    ret = []
    for each in cls.get_requirements(target_os=target_os):
        is_ignored = False
        for ignored in cls.REQUIREMENTS_IGNORED_ON_UNINSTALL:
            if each.startswith(ignored):
                is_ignored = True
                break
        if not is_ignored:
            ret.append(each)
    return ret

plugin_flavors() -> List[Type[BasePluginFlavor]] classmethod

Abstract method to declare new plugin flavors.

Returns:

Type	Description
List[Type[BasePluginFlavor]]	A list of new plugin flavors.

Source code in src/zenml/integrations/integration.py

@classmethod
def plugin_flavors(cls) -> List[Type["BasePluginFlavor"]]:
    """Abstract method to declare new plugin flavors.

    Returns:
        A list of new plugin flavors.
    """
    return []

Modules

data_validators

Initialization of the Deepchecks data validator for ZenML.

Classes

DeepchecksDataValidator(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: BaseDataValidator

Deepchecks data validator stack component.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Functions

data_validation(dataset: Union[pd.DataFrame, DataLoader[Any]], comparison_dataset: Optional[Any] = None, check_list: Optional[Sequence[str]] = None, dataset_kwargs: Dict[str, Any] = {}, check_kwargs: Dict[str, Dict[str, Any]] = {}, run_kwargs: Dict[str, Any] = {}, **kwargs: Any) -> SuiteResult

Run one or more Deepchecks data validation checks on a dataset.

Call this method to analyze and identify potential integrity problems with a single dataset (e.g. missing values, conflicting labels, mixed data types etc.) and dataset comparison checks (e.g. data drift checks). Dataset comparison checks require that a second dataset be supplied via the comparison_dataset argument.

The check_list argument may be used to specify a custom set of Deepchecks data integrity checks to perform, identified by DeepchecksDataIntegrityCheck and DeepchecksDataDriftCheck enum values. If omitted:

• if the comparison_dataset is omitted, a suite with all available data integrity checks will be performed on the input data. See DeepchecksDataIntegrityCheck for a list of Deepchecks builtin checks that are compatible with this method.

• if the comparison_dataset is supplied, a suite with all available data drift checks will be performed on the input data. See DeepchecksDataDriftCheck for a list of Deepchecks builtin checks that are compatible with this method.

Parameters:

Name	Type	Description	Default
dataset	Union[DataFrame, DataLoader[Any]]	Target dataset to be validated.	required
comparison_dataset	Optional[Any]	Optional second dataset to be used for data comparison checks (e.g data drift checks).	None
check_list	Optional[Sequence[str]]	Optional list of ZenML Deepchecks check identifiers specifying the data validation checks to be performed. DeepchecksDataIntegrityCheck enum values should be used for single data validation checks and DeepchecksDataDriftCheck enum values for data comparison checks. If not supplied, the entire set of checks applicable to the input dataset(s) will be performed.	None
dataset_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	{}
check_kwargs	Dict[str, Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	{}
run_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	{}
kwargs	Any	Additional keyword arguments (unused).	{}

Returns:

Type	Description
SuiteResult	A Deepchecks SuiteResult with the results of the validation.

Source code in src/zenml/integrations/deepchecks/data_validators/deepchecks_data_validator.py

def data_validation(
    self,
    dataset: Union[pd.DataFrame, DataLoader[Any]],
    comparison_dataset: Optional[Any] = None,
    check_list: Optional[Sequence[str]] = None,
    dataset_kwargs: Dict[str, Any] = {},
    check_kwargs: Dict[str, Dict[str, Any]] = {},
    run_kwargs: Dict[str, Any] = {},
    **kwargs: Any,
) -> SuiteResult:
    """Run one or more Deepchecks data validation checks on a dataset.

    Call this method to analyze and identify potential integrity problems
    with a single dataset (e.g. missing values, conflicting labels, mixed
    data types etc.) and dataset comparison checks (e.g. data drift
    checks). Dataset comparison checks require that a second dataset be
    supplied via the `comparison_dataset` argument.

    The `check_list` argument may be used to specify a custom set of
    Deepchecks data integrity checks to perform, identified by
    `DeepchecksDataIntegrityCheck` and `DeepchecksDataDriftCheck` enum
    values. If omitted:

    * if the `comparison_dataset` is omitted, a suite with all available
    data integrity checks will be performed on the input data. See
    `DeepchecksDataIntegrityCheck` for a list of Deepchecks builtin
    checks that are compatible with this method.

    * if the `comparison_dataset` is supplied, a suite with all
    available data drift checks will be performed on the input
    data. See `DeepchecksDataDriftCheck` for a list of Deepchecks
    builtin checks that are compatible with this method.

    Args:
        dataset: Target dataset to be validated.
        comparison_dataset: Optional second dataset to be used for data
            comparison checks (e.g data drift checks).
        check_list: Optional list of ZenML Deepchecks check identifiers
            specifying the data validation checks to be performed.
            `DeepchecksDataIntegrityCheck` enum values should be used for
            single data validation checks and `DeepchecksDataDriftCheck`
            enum values for data comparison checks. If not supplied, the
            entire set of checks applicable to the input dataset(s)
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.
        kwargs: Additional keyword arguments (unused).

    Returns:
        A Deepchecks SuiteResult with the results of the validation.
    """
    check_enum: Type[DeepchecksValidationCheck]
    if comparison_dataset is None:
        check_enum = DeepchecksDataIntegrityCheck
    else:
        check_enum = DeepchecksDataDriftCheck

    return self._create_and_run_check_suite(
        check_enum=check_enum,
        reference_dataset=dataset,
        comparison_dataset=comparison_dataset,
        check_list=check_list,
        dataset_kwargs=dataset_kwargs,
        check_kwargs=check_kwargs,
        run_kwargs=run_kwargs,
    )

model_validation(dataset: Union[pd.DataFrame, DataLoader[Any]], model: Union[ClassifierMixin, Module], comparison_dataset: Optional[Any] = None, check_list: Optional[Sequence[str]] = None, dataset_kwargs: Dict[str, Any] = {}, check_kwargs: Dict[str, Dict[str, Any]] = {}, run_kwargs: Dict[str, Any] = {}, **kwargs: Any) -> Any

Run one or more Deepchecks model validation checks.

Call this method to perform model validation checks (e.g. confusion matrix validation, performance reports, model error analyzes, etc). A second dataset is required for model performance comparison tests (i.e. tests that identify changes in a model behavior by comparing how it performs on two different datasets).

The check_list argument may be used to specify a custom set of Deepchecks model validation checks to perform, identified by DeepchecksModelValidationCheck and DeepchecksModelDriftCheck enum values. If omitted:

* if the `comparison_dataset` is omitted, a suite with all available
model validation checks will be performed on the input data. See
`DeepchecksModelValidationCheck` for a list of Deepchecks builtin
checks that are compatible with this method.

* if the `comparison_dataset` is supplied, a suite with all
available model comparison checks will be performed on the input
data. See `DeepchecksModelValidationCheck` for a list of Deepchecks
builtin checks that are compatible with this method.

Parameters:

Name	Type	Description	Default
dataset	Union[DataFrame, DataLoader[Any]]	Target dataset to be validated.	required
model	Union[ClassifierMixin, Module]	Target model to be validated.	required
comparison_dataset	Optional[Any]	Optional second dataset to be used for model comparison checks.	None
check_list	Optional[Sequence[str]]	Optional list of ZenML Deepchecks check identifiers specifying the model validation checks to be performed. DeepchecksModelValidationCheck enum values should be used for model validation checks and DeepchecksModelDriftCheck enum values for model comparison checks. If not supplied, the entire set of checks applicable to the input dataset(s) will be performed.	None
dataset_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	{}
check_kwargs	Dict[str, Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	{}
run_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	{}
kwargs	Any	Additional keyword arguments (unused).	{}

Returns:

Type	Description
Any	A Deepchecks SuiteResult with the results of the validation.

Source code in src/zenml/integrations/deepchecks/data_validators/deepchecks_data_validator.py

def model_validation(
    self,
    dataset: Union[pd.DataFrame, DataLoader[Any]],
    model: Union[ClassifierMixin, Module],
    comparison_dataset: Optional[Any] = None,
    check_list: Optional[Sequence[str]] = None,
    dataset_kwargs: Dict[str, Any] = {},
    check_kwargs: Dict[str, Dict[str, Any]] = {},
    run_kwargs: Dict[str, Any] = {},
    **kwargs: Any,
) -> Any:
    """Run one or more Deepchecks model validation checks.

    Call this method to perform model validation checks (e.g. confusion
    matrix validation, performance reports, model error analyzes, etc).
    A second dataset is required for model performance comparison tests
    (i.e. tests that identify changes in a model behavior by comparing how
    it performs on two different datasets).

    The `check_list` argument may be used to specify a custom set of
    Deepchecks model validation checks to perform, identified by
    `DeepchecksModelValidationCheck` and `DeepchecksModelDriftCheck` enum
    values. If omitted:

        * if the `comparison_dataset` is omitted, a suite with all available
        model validation checks will be performed on the input data. See
        `DeepchecksModelValidationCheck` for a list of Deepchecks builtin
        checks that are compatible with this method.

        * if the `comparison_dataset` is supplied, a suite with all
        available model comparison checks will be performed on the input
        data. See `DeepchecksModelValidationCheck` for a list of Deepchecks
        builtin checks that are compatible with this method.

    Args:
        dataset: Target dataset to be validated.
        model: Target model to be validated.
        comparison_dataset: Optional second dataset to be used for model
            comparison checks.
        check_list: Optional list of ZenML Deepchecks check identifiers
            specifying the model validation checks to be performed.
            `DeepchecksModelValidationCheck` enum values should be used for
            model validation checks and `DeepchecksModelDriftCheck` enum
            values for model comparison checks. If not supplied, the
            entire set of checks applicable to the input dataset(s)
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks tabular.Dataset or vision.VisionData constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.
        kwargs: Additional keyword arguments (unused).

    Returns:
        A Deepchecks SuiteResult with the results of the validation.
    """
    check_enum: Type[DeepchecksValidationCheck]
    if comparison_dataset is None:
        check_enum = DeepchecksModelValidationCheck
    else:
        check_enum = DeepchecksModelDriftCheck

    return self._create_and_run_check_suite(
        check_enum=check_enum,
        reference_dataset=dataset,
        comparison_dataset=comparison_dataset,
        models=[model],
        check_list=check_list,
        dataset_kwargs=dataset_kwargs,
        check_kwargs=check_kwargs,
        run_kwargs=run_kwargs,
    )

Modules

deepchecks_data_validator

Implementation of the Deepchecks data validator.

Classes

DeepchecksDataValidator(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: BaseDataValidator

Deepchecks data validator stack component.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Functions

data_validation(dataset: Union[pd.DataFrame, DataLoader[Any]], comparison_dataset: Optional[Any] = None, check_list: Optional[Sequence[str]] = None, dataset_kwargs: Dict[str, Any] = {}, check_kwargs: Dict[str, Dict[str, Any]] = {}, run_kwargs: Dict[str, Any] = {}, **kwargs: Any) -> SuiteResult

Run one or more Deepchecks data validation checks on a dataset.

Call this method to analyze and identify potential integrity problems with a single dataset (e.g. missing values, conflicting labels, mixed data types etc.) and dataset comparison checks (e.g. data drift checks). Dataset comparison checks require that a second dataset be supplied via the comparison_dataset argument.

The check_list argument may be used to specify a custom set of Deepchecks data integrity checks to perform, identified by DeepchecksDataIntegrityCheck and DeepchecksDataDriftCheck enum values. If omitted:

• if the comparison_dataset is omitted, a suite with all available data integrity checks will be performed on the input data. See DeepchecksDataIntegrityCheck for a list of Deepchecks builtin checks that are compatible with this method.

• if the comparison_dataset is supplied, a suite with all available data drift checks will be performed on the input data. See DeepchecksDataDriftCheck for a list of Deepchecks builtin checks that are compatible with this method.

Parameters:

Name	Type	Description	Default
dataset	Union[DataFrame, DataLoader[Any]]	Target dataset to be validated.	required
comparison_dataset	Optional[Any]	Optional second dataset to be used for data comparison checks (e.g data drift checks).	None
check_list	Optional[Sequence[str]]	Optional list of ZenML Deepchecks check identifiers specifying the data validation checks to be performed. DeepchecksDataIntegrityCheck enum values should be used for single data validation checks and DeepchecksDataDriftCheck enum values for data comparison checks. If not supplied, the entire set of checks applicable to the input dataset(s) will be performed.	None
dataset_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	{}
check_kwargs	Dict[str, Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	{}
run_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	{}
kwargs	Any	Additional keyword arguments (unused).	{}

Returns:

Type	Description
SuiteResult	A Deepchecks SuiteResult with the results of the validation.

Source code in src/zenml/integrations/deepchecks/data_validators/deepchecks_data_validator.py

def data_validation(
    self,
    dataset: Union[pd.DataFrame, DataLoader[Any]],
    comparison_dataset: Optional[Any] = None,
    check_list: Optional[Sequence[str]] = None,
    dataset_kwargs: Dict[str, Any] = {},
    check_kwargs: Dict[str, Dict[str, Any]] = {},
    run_kwargs: Dict[str, Any] = {},
    **kwargs: Any,
) -> SuiteResult:
    """Run one or more Deepchecks data validation checks on a dataset.

    Call this method to analyze and identify potential integrity problems
    with a single dataset (e.g. missing values, conflicting labels, mixed
    data types etc.) and dataset comparison checks (e.g. data drift
    checks). Dataset comparison checks require that a second dataset be
    supplied via the `comparison_dataset` argument.

    The `check_list` argument may be used to specify a custom set of
    Deepchecks data integrity checks to perform, identified by
    `DeepchecksDataIntegrityCheck` and `DeepchecksDataDriftCheck` enum
    values. If omitted:

    * if the `comparison_dataset` is omitted, a suite with all available
    data integrity checks will be performed on the input data. See
    `DeepchecksDataIntegrityCheck` for a list of Deepchecks builtin
    checks that are compatible with this method.

    * if the `comparison_dataset` is supplied, a suite with all
    available data drift checks will be performed on the input
    data. See `DeepchecksDataDriftCheck` for a list of Deepchecks
    builtin checks that are compatible with this method.

    Args:
        dataset: Target dataset to be validated.
        comparison_dataset: Optional second dataset to be used for data
            comparison checks (e.g data drift checks).
        check_list: Optional list of ZenML Deepchecks check identifiers
            specifying the data validation checks to be performed.
            `DeepchecksDataIntegrityCheck` enum values should be used for
            single data validation checks and `DeepchecksDataDriftCheck`
            enum values for data comparison checks. If not supplied, the
            entire set of checks applicable to the input dataset(s)
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.
        kwargs: Additional keyword arguments (unused).

    Returns:
        A Deepchecks SuiteResult with the results of the validation.
    """
    check_enum: Type[DeepchecksValidationCheck]
    if comparison_dataset is None:
        check_enum = DeepchecksDataIntegrityCheck
    else:
        check_enum = DeepchecksDataDriftCheck

    return self._create_and_run_check_suite(
        check_enum=check_enum,
        reference_dataset=dataset,
        comparison_dataset=comparison_dataset,
        check_list=check_list,
        dataset_kwargs=dataset_kwargs,
        check_kwargs=check_kwargs,
        run_kwargs=run_kwargs,
    )

model_validation(dataset: Union[pd.DataFrame, DataLoader[Any]], model: Union[ClassifierMixin, Module], comparison_dataset: Optional[Any] = None, check_list: Optional[Sequence[str]] = None, dataset_kwargs: Dict[str, Any] = {}, check_kwargs: Dict[str, Dict[str, Any]] = {}, run_kwargs: Dict[str, Any] = {}, **kwargs: Any) -> Any

Run one or more Deepchecks model validation checks.

Call this method to perform model validation checks (e.g. confusion matrix validation, performance reports, model error analyzes, etc). A second dataset is required for model performance comparison tests (i.e. tests that identify changes in a model behavior by comparing how it performs on two different datasets).

The check_list argument may be used to specify a custom set of Deepchecks model validation checks to perform, identified by DeepchecksModelValidationCheck and DeepchecksModelDriftCheck enum values. If omitted:

* if the `comparison_dataset` is omitted, a suite with all available
model validation checks will be performed on the input data. See
`DeepchecksModelValidationCheck` for a list of Deepchecks builtin
checks that are compatible with this method.

* if the `comparison_dataset` is supplied, a suite with all
available model comparison checks will be performed on the input
data. See `DeepchecksModelValidationCheck` for a list of Deepchecks
builtin checks that are compatible with this method.

Parameters:

Name	Type	Description	Default
dataset	Union[DataFrame, DataLoader[Any]]	Target dataset to be validated.	required
model	Union[ClassifierMixin, Module]	Target model to be validated.	required
comparison_dataset	Optional[Any]	Optional second dataset to be used for model comparison checks.	None
check_list	Optional[Sequence[str]]	Optional list of ZenML Deepchecks check identifiers specifying the model validation checks to be performed. DeepchecksModelValidationCheck enum values should be used for model validation checks and DeepchecksModelDriftCheck enum values for model comparison checks. If not supplied, the entire set of checks applicable to the input dataset(s) will be performed.	None
dataset_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	{}
check_kwargs	Dict[str, Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	{}
run_kwargs	Dict[str, Any]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	{}
kwargs	Any	Additional keyword arguments (unused).	{}

Returns:

Type	Description
Any	A Deepchecks SuiteResult with the results of the validation.

Source code in src/zenml/integrations/deepchecks/data_validators/deepchecks_data_validator.py

def model_validation(
    self,
    dataset: Union[pd.DataFrame, DataLoader[Any]],
    model: Union[ClassifierMixin, Module],
    comparison_dataset: Optional[Any] = None,
    check_list: Optional[Sequence[str]] = None,
    dataset_kwargs: Dict[str, Any] = {},
    check_kwargs: Dict[str, Dict[str, Any]] = {},
    run_kwargs: Dict[str, Any] = {},
    **kwargs: Any,
) -> Any:
    """Run one or more Deepchecks model validation checks.

    Call this method to perform model validation checks (e.g. confusion
    matrix validation, performance reports, model error analyzes, etc).
    A second dataset is required for model performance comparison tests
    (i.e. tests that identify changes in a model behavior by comparing how
    it performs on two different datasets).

    The `check_list` argument may be used to specify a custom set of
    Deepchecks model validation checks to perform, identified by
    `DeepchecksModelValidationCheck` and `DeepchecksModelDriftCheck` enum
    values. If omitted:

        * if the `comparison_dataset` is omitted, a suite with all available
        model validation checks will be performed on the input data. See
        `DeepchecksModelValidationCheck` for a list of Deepchecks builtin
        checks that are compatible with this method.

        * if the `comparison_dataset` is supplied, a suite with all
        available model comparison checks will be performed on the input
        data. See `DeepchecksModelValidationCheck` for a list of Deepchecks
        builtin checks that are compatible with this method.

    Args:
        dataset: Target dataset to be validated.
        model: Target model to be validated.
        comparison_dataset: Optional second dataset to be used for model
            comparison checks.
        check_list: Optional list of ZenML Deepchecks check identifiers
            specifying the model validation checks to be performed.
            `DeepchecksModelValidationCheck` enum values should be used for
            model validation checks and `DeepchecksModelDriftCheck` enum
            values for model comparison checks. If not supplied, the
            entire set of checks applicable to the input dataset(s)
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks tabular.Dataset or vision.VisionData constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.
        kwargs: Additional keyword arguments (unused).

    Returns:
        A Deepchecks SuiteResult with the results of the validation.
    """
    check_enum: Type[DeepchecksValidationCheck]
    if comparison_dataset is None:
        check_enum = DeepchecksModelValidationCheck
    else:
        check_enum = DeepchecksModelDriftCheck

    return self._create_and_run_check_suite(
        check_enum=check_enum,
        reference_dataset=dataset,
        comparison_dataset=comparison_dataset,
        models=[model],
        check_list=check_list,
        dataset_kwargs=dataset_kwargs,
        check_kwargs=check_kwargs,
        run_kwargs=run_kwargs,
    )

Functions

flavors

Deepchecks integration flavors.

Classes

DeepchecksDataValidatorFlavor

Bases: BaseDataValidatorFlavor

Flavor of the Deepchecks data validator.

Attributes

docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[DeepchecksDataValidator] property

Implementation class.

Returns:

Type	Description
Type[DeepchecksDataValidator]	The implementation class.

logo_url: str property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
str	The flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

Modules

deepchecks_data_validator_flavor

Deepchecks data validator flavor.

Classes

DeepchecksDataValidatorFlavor

Bases: BaseDataValidatorFlavor

Flavor of the Deepchecks data validator.

Attributes

docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[DeepchecksDataValidator] property

Implementation class.

Returns:

Type	Description
Type[DeepchecksDataValidator]	The implementation class.

logo_url: str property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
str	The flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

materializers

Deepchecks materializers.

Classes

Modules

deepchecks_dataset_materializer

Implementation of Deepchecks dataset materializer.

Classes

DeepchecksDatasetMaterializer(uri: str, artifact_store: Optional[BaseArtifactStore] = None)

Bases: PandasMaterializer

Materializer to read data to and from Deepchecks dataset.

Source code in src/zenml/integrations/pandas/materializers/pandas_materializer.py

def __init__(
    self, uri: str, artifact_store: Optional[BaseArtifactStore] = None
):
    """Define `self.data_path`.

    Args:
        uri: The URI where the artifact data is stored.
        artifact_store: The artifact store where the artifact data is stored.
    """
    super().__init__(uri, artifact_store)
    try:
        import pyarrow  # type: ignore # noqa

        self.pyarrow_exists = True
    except ImportError:
        self.pyarrow_exists = False
        logger.warning(
            "By default, the `PandasMaterializer` stores data as a "
            "`.csv` file. If you want to store data more efficiently, "
            "you can install `pyarrow` by running "
            "'`pip install pyarrow`'. This will allow `PandasMaterializer` "
            "to automatically store the data as a `.parquet` file instead."
        )
    finally:
        self.parquet_path = os.path.join(self.uri, PARQUET_FILENAME)
        self.csv_path = os.path.join(self.uri, CSV_FILENAME)

Functions

extract_metadata(dataset: Dataset) -> Dict[str, MetadataType]

Extract metadata from the given Dataset object.

Parameters:

Name	Type	Description	Default
dataset	Dataset	The Dataset object to extract metadata from.	required

Returns:

Type	Description
Dict[str, MetadataType]	The extracted metadata as a dictionary.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_dataset_materializer.py

def extract_metadata(self, dataset: Dataset) -> Dict[str, "MetadataType"]:
    """Extract metadata from the given `Dataset` object.

    Args:
        dataset: The `Dataset` object to extract metadata from.

    Returns:
        The extracted metadata as a dictionary.
    """
    return super().extract_metadata(dataset.data)

load(data_type: Type[Any]) -> Dataset

Reads pandas dataframes and creates deepchecks.Dataset from it.

Parameters:

Name	Type	Description	Default
data_type	Type[Any]	The type of the data to read.	required

Returns:

Type	Description
Dataset	A Deepchecks Dataset.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_dataset_materializer.py

def load(self, data_type: Type[Any]) -> Dataset:
    """Reads pandas dataframes and creates `deepchecks.Dataset` from it.

    Args:
        data_type: The type of the data to read.

    Returns:
        A Deepchecks Dataset.
    """
    df = super().load(data_type)
    return Dataset(df)

save(dataset: Dataset) -> None

Serializes pandas dataframe within a Dataset object.

Parameters:

Name	Type	Description	Default
dataset	Dataset	A deepchecks.Dataset object.	required

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_dataset_materializer.py

def save(self, dataset: Dataset) -> None:
    """Serializes pandas dataframe within a `Dataset` object.

    Args:
        dataset: A deepchecks.Dataset object.
    """
    super().save(dataset.data)

save_visualizations(dataset: Dataset) -> Dict[str, VisualizationType]

Saves visualizations for the given Deepchecks dataset.

Parameters:

Name	Type	Description	Default
dataset	Dataset	The Deepchecks dataset to save visualizations for.	required

Returns:

Type	Description
Dict[str, VisualizationType]	A dictionary of visualization URIs and their types.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_dataset_materializer.py

def save_visualizations(
    self, dataset: Dataset
) -> Dict[str, VisualizationType]:
    """Saves visualizations for the given Deepchecks dataset.

    Args:
        dataset: The Deepchecks dataset to save visualizations for.

    Returns:
        A dictionary of visualization URIs and their types.
    """
    return super().save_visualizations(dataset.data)

deepchecks_results_materializer

Implementation of Deepchecks suite results materializer.

Classes

DeepchecksResultMaterializer(uri: str, artifact_store: Optional[BaseArtifactStore] = None)

Bases: BaseMaterializer

Materializer to read data to and from CheckResult and SuiteResult objects.

Source code in src/zenml/materializers/base_materializer.py

def __init__(
    self, uri: str, artifact_store: Optional[BaseArtifactStore] = None
):
    """Initializes a materializer with the given URI.

    Args:
        uri: The URI where the artifact data will be stored.
        artifact_store: The artifact store used to store this artifact.
    """
    self.uri = uri
    self._artifact_store = artifact_store

Functions

extract_metadata(result: Union[CheckResult, SuiteResult]) -> Dict[str, MetadataType]

Extract metadata from the given Deepchecks result.

Parameters:

Name	Type	Description	Default
result	Union[CheckResult, SuiteResult]	The Deepchecks result to extract metadata from.	required

Returns:

Type	Description
Dict[str, MetadataType]	The extracted metadata as a dictionary.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_results_materializer.py

def extract_metadata(
    self, result: Union[CheckResult, SuiteResult]
) -> Dict[str, "MetadataType"]:
    """Extract metadata from the given Deepchecks result.

    Args:
        result: The Deepchecks result to extract metadata from.

    Returns:
        The extracted metadata as a dictionary.
    """
    if isinstance(result, CheckResult):
        return {
            "deepchecks_check_name": result.get_header(),
            "deepchecks_check_passed": result.passed_conditions(),
        }
    elif isinstance(result, SuiteResult):
        return {
            "deepchecks_suite_name": result.name,
            "deepchecks_suite_passed": result.passed(),
        }
    return {}

load(data_type: Type[Any]) -> Union[CheckResult, SuiteResult]

Reads a Deepchecks check or suite result from a serialized JSON file.

Parameters:

Name	Type	Description	Default
data_type	Type[Any]	The type of the data to read.	required

Returns:

Type	Description
Union[CheckResult, SuiteResult]	A Deepchecks CheckResult or SuiteResult.

Raises:

Type	Description
RuntimeError	if the input data type is not supported.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_results_materializer.py

def load(self, data_type: Type[Any]) -> Union[CheckResult, SuiteResult]:
    """Reads a Deepchecks check or suite result from a serialized JSON file.

    Args:
        data_type: The type of the data to read.

    Returns:
        A Deepchecks CheckResult or SuiteResult.

    Raises:
        RuntimeError: if the input data type is not supported.
    """
    filepath = os.path.join(self.uri, RESULTS_FILENAME)

    json_res = io_utils.read_file_contents_as_string(filepath)
    if data_type == SuiteResult:
        res = SuiteResult.from_json(json_res)
    elif data_type == CheckResult:
        res = CheckResult.from_json(json_res)
    else:
        raise RuntimeError(f"Unknown data type: {data_type}")
    return res

save(result: Union[CheckResult, SuiteResult]) -> None

Creates a JSON serialization for a CheckResult or SuiteResult.

Parameters:

Name	Type	Description	Default
result	Union[CheckResult, SuiteResult]	A Deepchecks CheckResult or SuiteResult.	required

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_results_materializer.py

def save(self, result: Union[CheckResult, SuiteResult]) -> None:
    """Creates a JSON serialization for a CheckResult or SuiteResult.

    Args:
        result: A Deepchecks CheckResult or SuiteResult.
    """
    filepath = os.path.join(self.uri, RESULTS_FILENAME)
    serialized_json = result.to_json(True)
    io_utils.write_file_contents_as_string(filepath, serialized_json)

save_visualizations(result: Union[CheckResult, SuiteResult]) -> Dict[str, VisualizationType]

Saves visualizations for the given Deepchecks result.

Parameters:

Name	Type	Description	Default
result	Union[CheckResult, SuiteResult]	The Deepchecks result to save visualizations for.	required

Returns:

Type	Description
Dict[str, VisualizationType]	A dictionary of visualization URIs and their types.

Source code in src/zenml/integrations/deepchecks/materializers/deepchecks_results_materializer.py

def save_visualizations(
    self, result: Union[CheckResult, SuiteResult]
) -> Dict[str, VisualizationType]:
    """Saves visualizations for the given Deepchecks result.

    Args:
        result: The Deepchecks result to save visualizations for.

    Returns:
        A dictionary of visualization URIs and their types.
    """
    visualization_path = os.path.join(self.uri, HTML_FILENAME)
    visualization_path = visualization_path.replace("\\", "/")
    with fileio.open(visualization_path, "w") as f:
        result.save_as_html(f)
    return {visualization_path: VisualizationType.HTML}

Modules

steps

Initialization of the Deepchecks Standard Steps.

Functions

Modules

deepchecks_data_drift

Implementation of the Deepchecks data drift validation step.

Classes Functions

deepchecks_data_drift_check_step(reference_dataset: pd.DataFrame, target_dataset: pd.DataFrame, check_list: Optional[Sequence[DeepchecksDataDriftCheck]] = None, dataset_kwargs: Optional[Dict[str, Any]] = None, check_kwargs: Optional[Dict[str, Any]] = None, run_kwargs: Optional[Dict[str, Any]] = None) -> SuiteResult

Run data drift checks on two pandas datasets using Deepchecks.

Parameters:

Name	Type	Description	Default
reference_dataset	DataFrame	Reference dataset for the data drift check.	required
target_dataset	DataFrame	Target dataset to be used for the data drift check.	required
check_list	Optional[Sequence[DeepchecksDataDriftCheck]]	Optional list of DeepchecksDataDriftCheck identifiers specifying the subset of Deepchecks data drift checks to be performed. If not supplied, the entire set of data drift checks will be performed.	None
dataset_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	None
check_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	None
run_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	None

Returns:

Type	Description
SuiteResult	A Deepchecks suite result with the validation results.

Source code in src/zenml/integrations/deepchecks/steps/deepchecks_data_drift.py

@step
def deepchecks_data_drift_check_step(
    reference_dataset: pd.DataFrame,
    target_dataset: pd.DataFrame,
    check_list: Optional[Sequence[DeepchecksDataDriftCheck]] = None,
    dataset_kwargs: Optional[Dict[str, Any]] = None,
    check_kwargs: Optional[Dict[str, Any]] = None,
    run_kwargs: Optional[Dict[str, Any]] = None,
) -> SuiteResult:
    """Run data drift checks on two pandas datasets using Deepchecks.

    Args:
        reference_dataset: Reference dataset for the data drift check.
        target_dataset: Target dataset to be used for the data drift check.
        check_list: Optional list of DeepchecksDataDriftCheck identifiers
            specifying the subset of Deepchecks data drift checks to be
            performed. If not supplied, the entire set of data drift checks will
            be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.

    Returns:
        A Deepchecks suite result with the validation results.
    """
    data_validator = cast(
        DeepchecksDataValidator,
        DeepchecksDataValidator.get_active_data_validator(),
    )

    return data_validator.data_validation(
        dataset=reference_dataset,
        comparison_dataset=target_dataset,
        check_list=check_list,
        dataset_kwargs=dataset_kwargs or {},
        check_kwargs=check_kwargs or {},
        run_kwargs=run_kwargs or {},
    )

deepchecks_data_integrity

Implementation of the Deepchecks data integrity validation step.

Classes Functions

deepchecks_data_integrity_check_step(dataset: pd.DataFrame, check_list: Optional[Sequence[DeepchecksDataIntegrityCheck]] = None, dataset_kwargs: Optional[Dict[str, Any]] = None, check_kwargs: Optional[Dict[str, Any]] = None, run_kwargs: Optional[Dict[str, Any]] = None) -> SuiteResult

Run data integrity checks on a pandas dataset using Deepchecks.

Parameters:

Name	Type	Description	Default
dataset	DataFrame	a Pandas DataFrame to validate	required
check_list	Optional[Sequence[DeepchecksDataIntegrityCheck]]	Optional list of DeepchecksDataIntegrityCheck identifiers specifying the subset of Deepchecks data integrity checks to be performed. If not supplied, the entire set of data integrity checks will be performed.	None
dataset_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	None
check_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	None
run_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	None

Returns:

Type	Description
SuiteResult	A Deepchecks suite result with the validation results.

Source code in src/zenml/integrations/deepchecks/steps/deepchecks_data_integrity.py

@step
def deepchecks_data_integrity_check_step(
    dataset: pd.DataFrame,
    check_list: Optional[Sequence[DeepchecksDataIntegrityCheck]] = None,
    dataset_kwargs: Optional[Dict[str, Any]] = None,
    check_kwargs: Optional[Dict[str, Any]] = None,
    run_kwargs: Optional[Dict[str, Any]] = None,
) -> SuiteResult:
    """Run data integrity checks on a pandas dataset using Deepchecks.

    Args:
        dataset: a Pandas DataFrame to validate
        check_list: Optional list of DeepchecksDataIntegrityCheck identifiers
            specifying the subset of Deepchecks data integrity checks to be
            performed. If not supplied, the entire set of data integrity checks
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.

    Returns:
        A Deepchecks suite result with the validation results.
    """
    data_validator = cast(
        DeepchecksDataValidator,
        DeepchecksDataValidator.get_active_data_validator(),
    )

    return data_validator.data_validation(
        dataset=dataset,
        check_list=cast(Optional[Sequence[str]], check_list),
        dataset_kwargs=dataset_kwargs or {},
        check_kwargs=check_kwargs or {},
        run_kwargs=run_kwargs or {},
    )

deepchecks_model_drift

Implementation of the Deepchecks model drift validation step.

Classes Functions

deepchecks_model_drift_check_step(reference_dataset: pd.DataFrame, target_dataset: pd.DataFrame, model: ClassifierMixin, check_list: Optional[Sequence[DeepchecksModelDriftCheck]] = None, dataset_kwargs: Optional[Dict[str, Any]] = None, check_kwargs: Optional[Dict[str, Any]] = None, run_kwargs: Optional[Dict[str, Any]] = None) -> SuiteResult

Run model drift checks on two pandas DataFrames and an sklearn model.

Parameters:

Name	Type	Description	Default
reference_dataset	DataFrame	Reference dataset for the model drift check.	required
target_dataset	DataFrame	Target dataset to be used for the model drift check.	required
model	ClassifierMixin	a scikit-learn model to validate	required
check_list	Optional[Sequence[DeepchecksModelDriftCheck]]	Optional list of DeepchecksDataIntegrityCheck identifiers specifying the subset of Deepchecks data integrity checks to be performed. If not supplied, the entire set of data integrity checks will be performed.	None
dataset_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	None
check_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	None
run_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	None

Returns:

Type	Description
SuiteResult	A Deepchecks suite result with the validation results.

Source code in src/zenml/integrations/deepchecks/steps/deepchecks_model_drift.py

@step
def deepchecks_model_drift_check_step(
    reference_dataset: pd.DataFrame,
    target_dataset: pd.DataFrame,
    model: ClassifierMixin,
    check_list: Optional[Sequence[DeepchecksModelDriftCheck]] = None,
    dataset_kwargs: Optional[Dict[str, Any]] = None,
    check_kwargs: Optional[Dict[str, Any]] = None,
    run_kwargs: Optional[Dict[str, Any]] = None,
) -> SuiteResult:
    """Run model drift checks on two pandas DataFrames and an sklearn model.

    Args:
        reference_dataset: Reference dataset for the model drift check.
        target_dataset: Target dataset to be used for the model drift check.
        model: a scikit-learn model to validate
        check_list: Optional list of DeepchecksDataIntegrityCheck identifiers
            specifying the subset of Deepchecks data integrity checks to be
            performed. If not supplied, the entire set of data integrity checks
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.

    Returns:
        A Deepchecks suite result with the validation results.
    """
    data_validator = cast(
        DeepchecksDataValidator,
        DeepchecksDataValidator.get_active_data_validator(),
    )

    return data_validator.model_validation(
        dataset=reference_dataset,
        comparison_dataset=target_dataset,
        model=model,
        check_list=cast(Optional[Sequence[str]], check_list),
        dataset_kwargs=dataset_kwargs or {},
        check_kwargs=check_kwargs or {},
        run_kwargs=run_kwargs or {},
    )

deepchecks_model_validation

Implementation of the Deepchecks model validation validation step.

Classes Functions

deepchecks_model_validation_check_step(dataset: pd.DataFrame, model: ClassifierMixin, check_list: Optional[Sequence[DeepchecksModelValidationCheck]] = None, dataset_kwargs: Optional[Dict[str, Any]] = None, check_kwargs: Optional[Dict[str, Any]] = None, run_kwargs: Optional[Dict[str, Any]] = None) -> SuiteResult

Run model validation checks on a pandas DataFrame and an sklearn model.

Parameters:

Name	Type	Description	Default
dataset	DataFrame	a Pandas DataFrame to use for the validation	required
model	ClassifierMixin	a scikit-learn model to validate	required
check_list	Optional[Sequence[DeepchecksModelValidationCheck]]	Optional list of DeepchecksModelValidationCheck identifiers specifying the subset of Deepchecks model validation checks to be performed. If not supplied, the entire set of model validation checks will be performed.	None
dataset_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks tabular.Dataset or vision.VisionData constructor.	None
check_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks check object constructors. Arguments are grouped for each check and indexed using the full check class name or check enum value as dictionary keys.	None
run_kwargs	Optional[Dict[str, Any]]	Additional keyword arguments to be passed to the Deepchecks Suite run method.	None

Returns:

Type	Description
SuiteResult	A Deepchecks suite result with the validation results.

Source code in src/zenml/integrations/deepchecks/steps/deepchecks_model_validation.py

@step
def deepchecks_model_validation_check_step(
    dataset: pd.DataFrame,
    model: ClassifierMixin,
    check_list: Optional[Sequence[DeepchecksModelValidationCheck]] = None,
    dataset_kwargs: Optional[Dict[str, Any]] = None,
    check_kwargs: Optional[Dict[str, Any]] = None,
    run_kwargs: Optional[Dict[str, Any]] = None,
) -> SuiteResult:
    """Run model validation checks on a pandas DataFrame and an sklearn model.

    Args:
        dataset: a Pandas DataFrame to use for the validation
        model: a scikit-learn model to validate
        check_list: Optional list of DeepchecksModelValidationCheck identifiers
            specifying the subset of Deepchecks model validation checks to be
            performed. If not supplied, the entire set of model validation checks
            will be performed.
        dataset_kwargs: Additional keyword arguments to be passed to the
            Deepchecks `tabular.Dataset` or `vision.VisionData` constructor.
        check_kwargs: Additional keyword arguments to be passed to the
            Deepchecks check object constructors. Arguments are grouped for
            each check and indexed using the full check class name or
            check enum value as dictionary keys.
        run_kwargs: Additional keyword arguments to be passed to the
            Deepchecks Suite `run` method.

    Returns:
        A Deepchecks suite result with the validation results.
    """
    data_validator = cast(
        DeepchecksDataValidator,
        DeepchecksDataValidator.get_active_data_validator(),
    )

    return data_validator.model_validation(
        dataset=dataset,
        model=model,
        check_list=cast(Optional[Sequence[str]], check_list),
        dataset_kwargs=dataset_kwargs or {},
        check_kwargs=check_kwargs or {},
        run_kwargs=run_kwargs or {},
    )

validation_checks

Definition of the Deepchecks validation check types.

Classes

DeepchecksDataDriftCheck

Bases: DeepchecksValidationCheck

Categories of Deepchecks data drift checks.

This list reflects the set of train-test validation checks provided by Deepchecks:

• for tabular data
• for computer vision

All these checks inherit from deepchecks.tabular.TrainTestCheck or deepchecks.vision.TrainTestCheck and require two datasets as input.

DeepchecksDataIntegrityCheck

Bases: DeepchecksValidationCheck

Categories of Deepchecks data integrity checks.

This list reflects the set of data integrity checks provided by Deepchecks:

• for tabular data
• for computer vision

All these checks inherit from deepchecks.tabular.SingleDatasetCheck or deepchecks.vision.SingleDatasetCheck and require a single dataset as input.

DeepchecksModelDriftCheck

Bases: DeepchecksValidationCheck

Categories of Deepchecks model drift checks.

This list includes a subset of the model evaluation checks provided by Deepchecks that require two datasets and a mandatory model as input:

• for tabular data
• for computer vision

All these checks inherit from deepchecks.tabular.TrainTestCheck or deepchecks.vision.TrainTestCheck and require two datasets and a mandatory model as input.

DeepchecksModelValidationCheck

Bases: DeepchecksValidationCheck

Categories of Deepchecks model validation checks.

This list includes a subset of the model evaluation checks provided by Deepchecks that require a single dataset and a mandatory model as input:

• for tabular data
• for computer vision

All these checks inherit from deepchecks.tabular.SingleDatasetCheck or `deepchecks.vision.SingleDatasetCheck and require a dataset and a mandatory model as input.

DeepchecksValidationCheck

Bases: StrEnum

Base class for all Deepchecks categories of validation checks.

This base class defines some conventions used for all enum values used to identify the various validation checks that can be performed with Deepchecks:

• enum values represent fully formed class paths pointing to Deepchecks BaseCheck subclasses
• all tabular data checks are located under the deepchecks.tabular.checks module sub-tree
• all computer vision data checks are located under the deepchecks.vision.checks module sub-tree

Attributes

check_class: Type[BaseCheck] property

Convert the enum value to a valid Deepchecks check class.

Returns:

Type	Description
Type[BaseCheck]	The Deepchecks check class associated with the enum value.

Functions

get_check_class(check_name: str) -> Type[BaseCheck] classmethod

Get the Deepchecks check class associated with an enum value or a custom check name.

Parameters:

Name	Type	Description	Default
check_name	str	Identifies a builtin Deepchecks check. The identifier must be formatted as deepchecks.{tabular|vision}.checks.<class-name> and must be resolvable to a valid Deepchecks BaseCheck class.	required

Returns:

Type	Description
Type[BaseCheck]	The Deepchecks check class associated with this enum value.

Raises:

Type	Description
ValueError	If the check name could not be converted to a valid Deepchecks check class. This can happen for example if the enum values fall out of sync with the Deepchecks code base or if a custom check name is supplied that cannot be resolved to a valid Deepchecks BaseCheck class.

Source code in src/zenml/integrations/deepchecks/validation_checks.py

@classmethod
def get_check_class(cls, check_name: str) -> Type[BaseCheck]:
    """Get the Deepchecks check class associated with an enum value or a custom check name.

    Args:
        check_name: Identifies a builtin Deepchecks check. The identifier
            must be formatted as `deepchecks.{tabular|vision}.checks.<class-name>`
            and must be resolvable to a valid Deepchecks BaseCheck class.

    Returns:
        The Deepchecks check class associated with this enum value.

    Raises:
        ValueError: If the check name could not be converted to a valid
            Deepchecks check class. This can happen for example if the enum
            values fall out of sync with the Deepchecks code base or if a
            custom check name is supplied that cannot be resolved to a valid
            Deepchecks BaseCheck class.
    """
    cls.validate_check_name(check_name)

    try:
        check_class: Type[BaseCheck] = (
            source_utils.load_and_validate_class(
                check_name, expected_class=BaseCheck
            )
        )
    except AttributeError:
        raise ValueError(
            f"Could not map the `{check_name}` check identifier to a valid "
            f"Deepchecks check class."
        )

    if check_name not in cls.values():
        logger.warning(
            f"You are using a custom Deepchecks check identifier that is "
            f"not listed in the `{str(cls)}` enum type. This could lead "
            f"to unexpected behavior."
        )

    return check_class

is_tabular_check(check_name: str) -> bool classmethod

Check if a validation check is applicable to tabular data.

Parameters:

Name	Type	Description	Default
check_name	str	Identifies a builtin Deepchecks check.	required

Returns:

Type	Description
bool	True if the check is applicable to tabular data, otherwise False.

Source code in src/zenml/integrations/deepchecks/validation_checks.py

@classmethod
def is_tabular_check(cls, check_name: str) -> bool:
    """Check if a validation check is applicable to tabular data.

    Args:
        check_name: Identifies a builtin Deepchecks check.

    Returns:
        True if the check is applicable to tabular data, otherwise False.
    """
    cls.validate_check_name(check_name)
    return check_name.startswith("deepchecks.tabular.")

is_vision_check(check_name: str) -> bool classmethod

Check if a validation check is applicable to computer vision data.

Parameters:

Name	Type	Description	Default
check_name	str	Identifies a builtin Deepchecks check.	required

Returns:

Type	Description
bool	True if the check is applicable to compute vision data, otherwise
bool	False.

Source code in src/zenml/integrations/deepchecks/validation_checks.py

@classmethod
def is_vision_check(cls, check_name: str) -> bool:
    """Check if a validation check is applicable to computer vision data.

    Args:
        check_name: Identifies a builtin Deepchecks check.

    Returns:
        True if the check is applicable to compute vision data, otherwise
        False.
    """
    cls.validate_check_name(check_name)
    return check_name.startswith("deepchecks.vision.")

validate_check_name(check_name: str) -> None classmethod

Validate a Deepchecks check identifier.

Parameters:

Name	Type	Description	Default
check_name	str	Identifies a builtin Deepchecks check. The identifier must be formatted as deepchecks.{tabular|vision}.checks.<...>.<class-name>.	required

Raises:

Type	Description
ValueError	If the check identifier does not follow the convention used by ZenML to identify Deepchecks builtin checks.

Source code in src/zenml/integrations/deepchecks/validation_checks.py

@classmethod
def validate_check_name(cls, check_name: str) -> None:
    """Validate a Deepchecks check identifier.

    Args:
        check_name: Identifies a builtin Deepchecks check. The identifier
            must be formatted as `deepchecks.{tabular|vision}.checks.<...>.<class-name>`.

    Raises:
        ValueError: If the check identifier does not follow the convention
            used by ZenML to identify Deepchecks builtin checks.
    """
    if not re.match(
        r"^deepchecks\.(tabular|vision)\.checks\.",
        check_name,
    ):
        raise ValueError(
            f"The supplied Deepcheck check identifier does not follow the "
            f"convention used by ZenML: `{check_name}`. The identifier "
            f"must be formatted as `deepchecks.<tabular|vision>.checks...` "
            f"and must be resolvable to a valid Deepchecks BaseCheck "
            f"subclass."
        )

Functions

Modules