Repository: bigdatabr/deirokay (language: python)

Issue:
test: Unify tests for multibackend resources



Discussion / hints:

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 21ceb8e..d8be35b 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -2,12 +2,14 @@ repos:
   - repo: https://gitlab.com/pycqa/flake8
     rev: 4.0.1
     hooks:
-    - id: flake8
+      - id: flake8
+        args: ['deirokay', 'tests']
   - repo: https://github.com/pycqa/isort
     rev: 5.10.1
     hooks:
       - id: isort
         name: isort (python)
+        args: ['deirokay', 'tests']
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.1.0
     hooks:
diff --git a/deirokay/__init__.py b/deirokay/__init__.py
index 352354d..61acadd 100644
--- a/deirokay/__init__.py
+++ b/deirokay/__init__.py
@@ -1,8 +1,17 @@
 """
-Gather main Deirokay methods to be easile accessible `from deirokay`.
+Deirokay's main set of classes and functions.
 """
 
-from .parser.parser import data_reader
+from typing import TYPE_CHECKING
+
+if not TYPE_CHECKING:
+    import lazy_import
+    lazy_import.lazy_module('dask.dataframe')
+    lazy_import.lazy_module('numpy')
+    lazy_import.lazy_module('pandas')
+
+
+from .parser import data_reader
 from .profiler import profile
 from .validator import validate
 
diff --git a/deirokay/__version__.py b/deirokay/__version__.py
index 777f190..e1e430e 100644
--- a/deirokay/__version__.py
+++ b/deirokay/__version__.py
@@ -1,1 +1,2 @@
 __version__ = "0.8.0"
+__comp_version__ = tuple(int(i) for i in __version__.split('.'))
diff --git a/deirokay/_typing.py b/deirokay/_typing.py
index ad73340..f89de75 100644
--- a/deirokay/_typing.py
+++ b/deirokay/_typing.py
@@ -1,9 +1,48 @@
-from typing import Any, Dict, List, Union
+from typing import Any, Callable, Dict, List, TypeVar, Union
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+from typing_extensions import Literal, get_args
+
+try:
+    from typing_extensions import Protocol
+except ModuleNotFoundError:
+    # Protocol exists only since Python 3.8
+    Protocol = object
+
+from .enums import Backend
+
+DeirokayDataSource = TypeVar('DeirokayDataSource',
+                             'pandas.DataFrame', 'dask.dataframe.DataFrame')
+DeirokayDataSeries = TypeVar('DeirokayDataSeries',
+                             'pandas.Series', 'dask.dataframe.Series')
+
+
+# docstr-coverage:excused `No need for _typing module`
+class DeirokayReadCallable(Protocol):  # noqa: E302 # docstr-coverage comment
+    def __call__(self, data: Union[str, DeirokayDataSource],
+                 columns: List[str],
+                 *args: Any,
+                 **kwargs: Any) -> DeirokayDataSource: ...
+
+# TODO: assert if all backends are listed in TypeVars above
+
+
+BackendValue = Literal['pandas', 'dask']
+assert set(get_args(BackendValue)) == {backend.value for backend in Backend}
 
 DeirokayOption = Dict[str, Any]
-DeirokayOptionsDocument = Dict[str, Union[str, Dict[str, DeirokayOption]]]
+DeirokayColumnOptions = Dict[str, DeirokayOption]
+DeirokayOptionsDocument = Dict[str, Union[str, DeirokayColumnOptions]]
 
 DeirokayStatement = Dict[str, Any]
 DeirokayValidationItem = Dict[str, Union[str, List[DeirokayStatement]]]
 DeirokayValidationDocument = Dict[str,
                                   Union[str, List[DeirokayValidationItem]]]
+DeirokayValidationResultDocument = Dict
+
+DeirokaySerializedSeries = Dict[Literal['values', 'parser'], Union[List, Dict]]
+
+AnyClass = TypeVar('AnyClass', bound=type)
+AnyCallable = Callable[..., Any]
+GeneralDecorator = Callable[[AnyCallable], AnyCallable]
diff --git a/deirokay/_utils.py b/deirokay/_utils.py
new file mode 100644
index 0000000..1a4a3f4
--- /dev/null
+++ b/deirokay/_utils.py
@@ -0,0 +1,90 @@
+from difflib import get_close_matches
+from typing import Generator, Iterable, TypeVar, Union
+
+from jinja2.nativetypes import Environment
+
+C = TypeVar('C', bound=type)
+
+
+def recursive_subclass_generator(cls: C) -> Generator[C, None, None]:
+    """Iterate over all subclasses of `cls` recursively (including
+    itself).
+
+    Parameters
+    ----------
+    cls : C
+        Class to iterate over.
+
+    Yields
+    ------
+    Generator[C, None, None]
+        Each of its subclasses, recursively.
+    """
+    yield cls
+    for subclass in cls.__subclasses__():
+        yield from recursive_subclass_generator(subclass)
+
+
+def check_columns_in_df_columns(columns: Iterable,
+                                df_columns: Iterable) -> None:
+    """Raise an exception if any of the elements from `columns` is not
+    present in `df_columns`.
+
+    Parameters
+    ----------
+    columns : Iterable
+        List of requested elements.
+    df_columns : Iterable
+        List of available elements.
+
+    Raises
+    ------
+    KeyError
+        Element not found.
+    """
+    miss = {}
+    for col in columns:
+        if col not in df_columns:
+            match = get_close_matches(col, df_columns, n=1, cutoff=0)[0]
+            miss[col] = match
+    if miss:
+        raise KeyError(f'Columns {list(miss.keys())} not found in your data.'
+                       f' Did you mean {list(miss.values())}?')
+
+
+def render_list(env: Environment, list_: list, template: dict):
+    """Render Jinja templates in list recursively."""
+    for index, value in enumerate(list_):
+        if isinstance(value, str) and '{{' in value:
+            rendered = env.from_string(value).render(**template)
+            list_[index] = rendered
+        elif isinstance(value, dict):
+            render_dict(env, value, template)
+        elif isinstance(value, list):
+            render_list(env, value, template)
+
+
+def render_dict(env: Environment, dict_: dict, template: dict):
+    """Render Jinja templates in dict recursively.
+
+    It will render only strings starting with `{{` to prevent
+    unattended renderings."""
+    for key, value in dict_.items():
+        if isinstance(value, str) and '{{' in value:
+            rendered = env.from_string(value).render(**template)
+            dict_[key] = rendered
+        elif isinstance(value, dict):
+            render_dict(env, value, template)
+        elif isinstance(value, list):
+            render_list(env, value, template)
+
+
+T = TypeVar('T')
+
+
+def noneor(*operands: T) -> Union[T, None]:
+    """Return the first non-None element."""
+    for item in operands:
+        if item is not None:
+            return item
+    return None
diff --git a/deirokay/airflow/__init__.py b/deirokay/airflow/__init__.py
index b264868..fa7a60e 100644
--- a/deirokay/airflow/__init__.py
+++ b/deirokay/airflow/__init__.py
@@ -1,5 +1,5 @@
 """
-Airflow extension using Deirokay.
+Airflow extensions using Deirokay.
 """
 
 from .deirokay_operator import DeirokayOperator
diff --git a/deirokay/airflow/deirokay_operator.py b/deirokay/airflow/deirokay_operator.py
index bdb89ff..b1f34ed 100644
--- a/deirokay/airflow/deirokay_operator.py
+++ b/deirokay/airflow/deirokay_operator.py
@@ -7,7 +7,9 @@
 from airflow.models.baseoperator import BaseOperator
 
 import deirokay
-from deirokay.enums import SeverityLevel
+from deirokay._typing import (DeirokayDataSource, DeirokayOptionsDocument,
+                              DeirokayValidationDocument)
+from deirokay.enums import Backend, SeverityLevel
 from deirokay.exceptions import ValidationError
 from deirokay.validator import raise_validation
 
@@ -33,6 +35,8 @@ class DeirokayOperator(BaseOperator):
     against : Union[dict, str]
         A dict or a local/S3 path to a YAML/JSON validation
         document file.
+    backend : Optional[Backend]
+        Backend to use when processing data.
     template : Optional[dict]
         Map of templates to be passed to Deirokay validation.
     save_to : Optional[str], optional
@@ -71,10 +75,12 @@ class DeirokayOperator(BaseOperator):
     def __init__(
         self,
         *,
-        data: Optional[str] = None,
-        path_to_file: Optional[str] = None,  # Deprecated
-        options: Union[dict, str],
-        against: Union[dict, str],
+        data: Optional[Union[str, DeirokayDataSource]] = None,
+        # `path_to_file` is Deprecated
+        path_to_file: Optional[Union[str, DeirokayDataSource]] = None,
+        options: Union[str, DeirokayOptionsDocument],
+        against: Union[str, DeirokayValidationDocument],
+        backend: Optional[Backend] = None,
         template: Optional[dict] = None,
         save_to: Optional[str] = None,
         soft_fail_level: Union[SeverityLevel, int] = SeverityLevel.MINIMAL,
@@ -94,11 +100,12 @@ def __init__(
                 ' removed in next major release. Use `data` instead.',
                 DeprecationWarning
             )
-        assert options
-        assert against
+        assert options, 'You should provide an `options` attribute.'
+        assert against, 'You should provide an `against` attribute.'
         self.data = data or path_to_file
         self.options = options
         self.against = against
+        self.backend = backend
         self.template = template
         self.save_to = save_to
         self.soft_fail_level = soft_fail_level
@@ -112,6 +119,7 @@ def execute(self, context: dict):
         df = deirokay.data_reader(
             self.data,
             options=self.options,
+            backend=self.backend,
             **self.reader_kwargs
         )
 
diff --git a/deirokay/backend.py b/deirokay/backend.py
new file mode 100644
index 0000000..2071736
--- /dev/null
+++ b/deirokay/backend.py
@@ -0,0 +1,289 @@
+import functools
+import warnings
+from typing import Callable, Dict, List, Optional, Type, TypeVar, Union
+
+from deirokay.exceptions import InvalidBackend, UnsupportedBackend
+
+from .__version__ import __comp_version__
+from ._typing import AnyCallable, DeirokayDataSource
+from .enums import Backend
+
+MODULE_2_BACKEND = {
+    'pandas': Backend.PANDAS,
+    'dask.dataframe': Backend.DASK,
+}
+assert sorted(Backend) == sorted(MODULE_2_BACKEND.values())
+
+BACKEND_2_MODULE = {v: k for k, v in MODULE_2_BACKEND.items()}
+
+
+_AnyMultiBackendClass = TypeVar('_AnyMultiBackendClass',
+                                bound='MultiBackendMixin')
+
+
+class MultiBackendMixin:
+    """Mixin class which all multi-backend resources in Deirokay
+    derives from.
+
+    Together with the `register_backend_method` decorator, the methods
+    of its subclasses can be marked to be active when a particular
+    backend is active.
+    """
+    supported_backends: List[Backend] = []
+    """List[Backend]: Backends supported by this resource."""
+    _current_backend: Optional[Backend] = None
+    """Optional[Backend]: Current active backend
+    (only valid during runtime)."""
+    _backend_methods: Dict[Backend, Dict[str, AnyCallable]] = {}
+
+    @classmethod
+    def register_backend_method(cls,
+                                alias_for: str,
+                                func: AnyCallable,
+                                backend: Backend) -> None:
+        """Proxy for `register_backend_method` to register an existing
+        function as a backend-specific method.
+
+        Parameters
+        ----------
+        alias_for : str
+            The name of the method to be substituted with a
+            backend-specific version.
+        func : AnyCallable
+            Existing function to be registered as a method.
+        backend : Backend
+            Backend for the method.
+        """
+        backend_method = register_backend_method(alias_for, backend)(func)
+        method_name = f'_registered_{alias_for}_{backend.value}'
+        setattr(cls, method_name, backend_method)
+        # When the decorator is not called from inside a class scope,
+        # the __set_name__ hook should be invoked manually.
+        # See: https://docs.python.org/3/reference/datamodel.html?highlight=__set_name__#object.__set_name__  # noqa: E501
+        backend_method.__set_name__(cls, method_name)  # type: ignore  # The decorator replaces itself in the owner class with the original method  # noqa: E501
+
+    @classmethod
+    @functools.lru_cache(maxsize=None)
+    def attach_backend(cls: Type['MultiBackendMixin'],
+                       backend: Backend
+                       ) -> Type['_AnyMultiBackendClass']:
+        """Generate a subclass that concretizes multibackend backend
+        methods into their intended name. The methods marked with the
+        given `backend` will compose the returned class.
+
+        Parameters
+        ----------
+        cls : type
+            Class to be subclassed with the given backend.
+        backend : Backend
+            Backend to be selected.
+
+        Returns
+        -------
+        Type[MultiBackendMixin]
+            Subclass of the current class with methods filtered for the
+            given backend.
+        """
+        if (
+            __comp_version__ < (2,)
+            and cls.supported_backends == []
+            and 'supported_backends' not in vars(cls)
+        ):
+            warnings.warn(
+                'To preserve backward compatibility, the'
+                f' `supported_backends` attribute from `{cls.__name__}`'
+                ' is assumed to be `[Backend.PANDAS]` when not declared.'
+                ' In future, this behavior will change and an exception'
+                ' will be raised whenever a multibackend class does not'
+                ' specify this attribute explicitely.\n'
+                'To prevent this error in future and suppress this'
+                ' warning in the current version, please set the'
+                ' `supported_backends` class attribute explicitely.',
+                FutureWarning
+            )
+            cls.supported_backends = [Backend.PANDAS]
+
+        if backend not in cls.supported_backends:
+            raise UnsupportedBackend(
+                f"The `{cls.__name__}` class does not support the '{backend}'"
+                f' backend, only the following ones: {cls.supported_backends}.'
+                f' If you are using a custom class, be sure to provide a'
+                f' `supported_backends` attribute containing {backend!s}.'
+            )
+
+        execution_name = f'{cls.__name__}-{backend.value.capitalize()}Backend'
+        execution_attrs = dict(vars(cls))
+        execution_subclass = type(
+            execution_name,
+            (cls,),
+            execution_attrs
+        )  # type: Type[_AnyMultiBackendClass]
+
+        _merged_backend_methods = {}
+        for _cls in reversed(cls.mro()):
+            if not issubclass(_cls, MultiBackendMixin):
+                continue
+            try:
+                _merged_backend_methods.update(_cls._backend_methods[backend])
+            except KeyError:
+                continue
+
+        for alias_for, (method, force) in _merged_backend_methods.items():
+            if not force and alias_for in vars(execution_subclass):
+                raise InvalidBackend(
+                    f'You cannot declare the alias method `{method.__name__}`'
+                    f' in the class `{cls}` when the original method'
+                    f' `{alias_for}` already exists or has already been'
+                    ' replaced by another alias.'
+                    f' Either remove `{alias_for}` or make sure only'
+                    f' one alias of `{alias_for}` exists for the'
+                    f" '{backend.value}' backend to continue.\n"
+                    "Alternatively, you may set the flag `force=True`"
+                    " when calling"
+                )
+            setattr(execution_subclass, alias_for, method)
+
+        # Save backend for future reference (via `get_backend`)
+        execution_subclass._current_backend = backend
+        # Call overwritable user method
+        execution_subclass.__post_attach_backend__()
+        return execution_subclass
+
+    @classmethod
+    def __post_attach_backend__(cls):
+        """This classmethod can be optionally overwritten to serve as a
+        callback function for when the `attach_backend()` method is
+        called."""
+
+    @classmethod
+    def get_backend(cls) -> Backend:
+        """Get current active backend for this class.
+
+        Returns
+        -------
+        Backend
+            The current active backend.
+
+        Raises
+        ------
+        InvalidBackend
+            Backend not set or not a valid execution class.
+        """
+        if cls._current_backend is None:
+            raise InvalidBackend('Backend not set for current execution')
+        return cls._current_backend
+
+
+DecoratorClass = Union[Callable, Type]
+
+
+def register_backend_method(
+    alias_for: str,
+    backend: Backend,
+    *,
+    force: bool = False
+) -> DecoratorClass:
+    """Modify a method to make it an alternative (alias) for another
+    method (`alias_for`) when the specified backend is active.
+    Should be used as an decorator as in:
+    `@register_backend_method('method_name', <Backend object>)`.
+
+    It could be handful to declare helper decorators using this one for
+    common methods for a given resource. For instance, if a hierarchy
+    of classes should implement a `do_it` method to work with different
+    backend, we could implement a `do_it` decorator and use it to
+    decorate backend-specific methods.
+
+    .. code-block:: python
+
+        def do_it(backend):
+            register_backend_method('do_it', backend)
+
+        class MultiBackendClass(MultiBackendMixin):
+            @do_it(Backend.TYPE1)
+            def _do_it_1(...):
+                ...
+
+            @do_it(Backend.TYPE2)
+            def _do_it_2(...):
+                ...
+
+    Parameters
+    ----------
+    alias_for : str
+        The name of the method to be substituted with a
+        backend-specific version.
+    backend : Backend
+        The backend to use when running the method.
+    force : bool, optional
+        Force overwrite target method when it already exists.
+        Defaults to False.
+    """
+    assert isinstance(backend, Backend), (
+        'Make sure this decorator declares a `backend` argument'
+    )
+
+    class _decorator():
+        def __init__(self, decorated_method: AnyCallable) -> None:
+            self.decorated_method = decorated_method
+
+        def __set_name__(self, owner: type, method_name: str) -> None:
+            setattr(owner, method_name, self.decorated_method)
+            if not issubclass(owner, MultiBackendMixin):
+                raise InvalidBackend(
+                    f'Make sure the `{owner.__name__}` class subclasses'
+                    f' `{MultiBackendMixin.__name__}` before using'
+                    f' `{register_backend_method.__name__}`'
+                )
+            if backend not in owner.supported_backends:
+                raise UnsupportedBackend(
+                    f'The `{owner.__name__}` class does not seem to support'
+                    f" the '{backend.value}' backend used in `{method_name}`."
+                    ' Make sure to explicitely declare a `supported_backends`'
+                    f" attribute containing the '{backend.value}' backend."
+                )
+            if '_backend_methods' not in vars(owner):
+                # We must reinitialize _backend_methods in all subclasses to
+                # prevent all of them sharing the same
+                # `MultiBackendMixin._backend_methods` attribute
+                owner._backend_methods = {}
+            if backend not in owner._backend_methods:
+                owner._backend_methods[backend] = {}
+            # We use `vars()` to get method since `getattr()` would resolve
+            # existing descriptors (notably, staticmethod and classmethod)
+            method = vars(owner)[method_name]
+            owner._backend_methods[backend][alias_for] = (method, force)
+
+    return _decorator
+
+
+def _detect_backend_from_type(object_type: type) -> Backend:
+    """Infers the backend from the object's module name."""
+    for _module in MODULE_2_BACKEND:
+        if object_type.__module__.startswith(_module):
+            return MODULE_2_BACKEND[_module]
+    raise UnsupportedBackend(
+        f'Unknown backend for {object_type}.'
+        f' Supported backends: {set(i.value for i in Backend)}.'
+    )
+
+
+def detect_backend(df: DeirokayDataSource) -> Backend:
+    """Map the object class to the proper backend value.
+
+    Parameters
+    ----------
+    df : Union[DataFrame, DaskDataFrame]
+        DataFrame object.
+
+    Returns
+    -------
+    Backend
+        Instance of `Backend` enumeration.
+
+    Raises
+    ------
+    ValueError
+        Unknown backend for the specified object type.
+    """
+    return _detect_backend_from_type(type(df))
diff --git a/deirokay/enums.py b/deirokay/enums.py
index 68d8764..be67804 100644
--- a/deirokay/enums.py
+++ b/deirokay/enums.py
@@ -2,7 +2,16 @@
 Deirokay enumeration classes.
 """
 
-from enum import Enum
+from enum import Enum, IntEnum, unique
+
+
+@unique
+class Backend(str, Enum):
+    """
+    The backend that the statement will be generated for.
+    """
+    PANDAS = 'pandas'
+    DASK = 'dask'
 
 
 class DTypes(str, Enum):
@@ -44,7 +53,8 @@ class DTypes(str, Enum):
     :ref:`DecimalTreater<deirokay.parser.treaters.DecimalTreater>`."""
 
 
-class SeverityLevel(int, Enum):
+@unique
+class SeverityLevel(IntEnum):
     """Deirokay named Severity levels."""
 
     MINIMAL = 1
diff --git a/deirokay/exceptions.py b/deirokay/exceptions.py
index 308bd01..773d91d 100644
--- a/deirokay/exceptions.py
+++ b/deirokay/exceptions.py
@@ -11,3 +11,19 @@ class ValidationError(Exception):
     def __init__(self, level: SeverityLevel, message='Validation failed'):
         self.level = level
         super().__init__(message)
+
+
+class InvalidBackend(ValueError):
+    """Error in backend definition."""
+
+
+class UnsupportedBackend(InvalidBackend):
+    """Backend is not supported for this resource."""
+
+
+class ParsingError(RuntimeError):
+    """Error during file parsing or object conversion."""
+
+
+class InvalidStatement(SyntaxError):
+    """Error when validating statement construction."""
diff --git a/deirokay/fs.py b/deirokay/fs.py
index 34b8ad1..e42f174 100644
--- a/deirokay/fs.py
+++ b/deirokay/fs.py
@@ -15,9 +15,10 @@
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from types import ModuleType
-from typing import IO, Optional, Sequence
+from typing import IO, Generator, Iterable, Optional, Sequence
 
 import yaml
+from typing_extensions import Literal
 
 boto3_import_error = None
 try:
@@ -211,13 +212,13 @@ def import_as_python_module(self) -> ModuleType:
         """Import file as a Python module."""
         raise NotImplementedError
 
-    def open(self, *args, **kwargs) -> IO:
+    def open(self, mode: Literal['r', 'w'], *args, **kwargs) -> IO:
         """Open file."""
         raise NotImplementedError
 
     def read(self, *args, **kwargs) -> str:
         """Read a file as text."""
-        with self.open(*args, **kwargs) as fp:
+        with self.open(*args, mode='r', **kwargs) as fp:
             return fp.read()
 
     def __truediv__(self, rest: str) -> 'FileSystem':
@@ -289,8 +290,8 @@ def mkdir(self, *args, **kwargs) -> None:
         return Path(self.path).mkdir(*args, **kwargs)
 
     # docstr-coverage:inherited
-    def open(self, *args, **kwargs) -> IO:
-        return open(self.path, *args, **kwargs)
+    def open(self, mode: Literal['r', 'w'], *args, **kwargs) -> IO:
+        return open(self.path, mode, *args, **kwargs)
 
 
 class S3FileSystem(FileSystem):
@@ -342,6 +343,7 @@ def ls(self, recursive: bool = False, files_only: bool = False,
             Prefix=self.prefix_or_key,
             PaginationConfig={'PageSize': max_keys}
         )
+        chained_items: Iterable
         if not reverse:
             chained_items = (
                 item['Key']
@@ -384,13 +386,13 @@ def import_as_python_module(self) -> ModuleType:
             return module
 
     # docstr-coverage:inherited
-    def open(self, mode, *args, **kwargs) -> IO:
+    def open(self, mode: Literal['r', 'w'], *args, **kwargs) -> IO:
         if mode == 'r':
             body = self.client.get_object(Bucket=self.bucket,
                                           Key=self.prefix_or_key)['Body']
-            return contextlib.closing(body)
+            return contextlib.closing(body)  # type: ignore
         elif mode == 'w':
-            def _writable():
+            def _writable() -> Generator[IO, None, None]:
                 with NamedTemporaryFile(mode, *args, **kwargs) as tmp_fp:
                     yield tmp_fp
                     tmp_fp.flush()
@@ -399,7 +401,7 @@ def _writable():
                         self.bucket,
                         self.prefix_or_key
                     )
-            return contextlib.contextmanager(_writable)()
+            return contextlib.contextmanager(_writable)()  # type: ignore
         raise NotImplementedError(f"Mode '{mode}' not supported.")
 
 
diff --git a/deirokay/history_template.py b/deirokay/history_template.py
index eee8f95..d794bdb 100644
--- a/deirokay/history_template.py
+++ b/deirokay/history_template.py
@@ -7,7 +7,7 @@
 from typing import Callable, List
 
 import jq
-from pandas import Series
+import pandas  # lazy module
 
 from ._typing import DeirokayValidationDocument
 from .fs import FileSystem
@@ -59,7 +59,7 @@ def __init__(self, statements) -> None:
                 jq.compile(f'.[].report.detail["{att}"]')
                 .input(statements).all()
             )
-            setattr(self, att, Series(child))
+            setattr(self, att, pandas.Series(child))
 
     def __getattr__(self, name: str) -> NullCallableNode:
         return NullCallableNode()
diff --git a/deirokay/parser/__init__.py b/deirokay/parser/__init__.py
index faf3dc6..092e93f 100644
--- a/deirokay/parser/__init__.py
+++ b/deirokay/parser/__init__.py
@@ -1,6 +1,12 @@
-from .parser import get_dtype_treater, get_treater_instance
+"""
+Classes and methods to load and parse correct data into Deirokay.
+"""
+
+from .loader import data_reader
+from .treaters import get_dtype_treater, get_treater_instance
 
 __all__ = (
+    'data_reader',
     'get_dtype_treater',
     'get_treater_instance',
 )
diff --git a/deirokay/parser/loader.py b/deirokay/parser/loader.py
new file mode 100644
index 0000000..be1bb5d
--- /dev/null
+++ b/deirokay/parser/loader.py
@@ -0,0 +1,107 @@
+"""
+Functions to parse files into pandas DataFrames.
+"""
+import warnings
+from copy import deepcopy
+from typing import Optional, Union
+
+from deirokay.__version__ import __comp_version__
+from deirokay._typing import (DeirokayColumnOptions, DeirokayDataSource,
+                              DeirokayOptionsDocument)
+from deirokay._utils import check_columns_in_df_columns
+from deirokay.backend import detect_backend
+from deirokay.enums import Backend
+from deirokay.exceptions import InvalidBackend, ParsingError
+from deirokay.fs import fs_factory
+
+from .reader import reader_factory
+from .treaters import get_treater_instance
+
+
+def data_reader(data: Union[str, DeirokayDataSource],
+                options: Union[str, DeirokayOptionsDocument],
+                backend: Optional[Backend] = None,
+                **kwargs) -> DeirokayDataSource:
+    """Create a new tabular data from a file or an object and apply
+    Deirokay treatments to correctly parse it and pre-validate its
+    content.
+
+    Parameters
+    ----------
+    data : Union[str, DeirokayDataSource]
+        Path or object used as data source.
+    options : Union[str, DeirokayOptionsDocument]
+        Path to or `dict` representing a Deirokay Options document.
+    backend: Optional[BackendValue], optional
+        Defines backend to use for tables. By default None. Inferred
+        from `data` when it is a valid DeirokayDataSource. Should be
+        set when `data` is a path.
+
+    Returns
+    -------
+    DeirokayDataSource
+        A tabular data treated by Deirokay.
+    """
+    if isinstance(options, str):
+        options_dict = fs_factory(options).read_dict()
+    else:
+        options_dict = deepcopy(options)
+    options_dict.update(kwargs)
+    columns = options_dict.pop('columns')
+
+    if isinstance(data, str) and backend is None:
+        if __comp_version__ < (2,):
+            warnings.warn(
+                'To preserve backward compatibility, the `backend` attribute'
+                ' is assumed to be `Backend.PANDAS` when reading data from a'
+                ' file or SQL query.\n'
+                'In future, this behavior will change and an exception will'
+                ' be raised whenever the backend cannot be inferred from the'
+                ' `data` attribute. To prevent this error in future and'
+                ' suppress this warning in the current version,'
+                ' set the `backend` attribute explicitely in `data_reader()`.',
+                FutureWarning
+            )
+            backend = Backend.PANDAS
+        elif __comp_version__ >= (2,):
+            raise InvalidBackend(
+                'You should provide a `backend` attribute when it cannot be'
+                ' inferred from `data`.'
+            )
+    backend = backend or detect_backend(data)
+
+    reader = reader_factory(backend)
+
+    df = reader.read(data, columns=list(columns), **options_dict)
+    data_treater(df, columns, backend)
+
+    return df
+
+
+def data_treater(df: DeirokayDataSource,
+                 options: DeirokayColumnOptions,
+                 backend: Backend) -> None:
+    """Receive options dict and call the proper treater class for each
+    Deirokay data type.
+
+    Parameters
+    ----------
+    df : DeirokayDataSource
+        Raw tabular data to be treated.
+    options : DeirokayColumnOptions
+        Deirokay options.
+
+    Raises
+    ------
+    NotImplementedError
+        Data type not valid or not implemented.
+    """
+    check_columns_in_df_columns(options.keys(), df.columns)
+
+    for col, opt in options.items():
+        option: dict = opt.copy()
+
+        try:
+            df[col] = get_treater_instance(option, backend)(df[col])
+        except Exception as e:
+            raise ParsingError(f'Error when parsing "{col}".') from e
diff --git a/deirokay/parser/parser.py b/deirokay/parser/parser.py
deleted file mode 100644
index 5819d6c..0000000
--- a/deirokay/parser/parser.py
+++ /dev/null
@@ -1,214 +0,0 @@
-"""
-Functions to parse files into pandas DataFrames.
-"""
-
-import datetime
-import decimal
-from os.path import splitext
-from typing import Any, Dict, List, Type, Union
-
-import pandas
-from pandas import (DataFrame, Timestamp, read_csv, read_excel, read_parquet,
-                    read_sql)
-
-from deirokay._typing import DeirokayOption, DeirokayOptionsDocument
-from deirokay.enums import DTypes
-from deirokay.fs import fs_factory
-from deirokay.utils import _check_columns_in_df_columns
-
-from . import treaters
-
-
-def data_reader(data: Union[str, DataFrame],
-                options: Union[str, DeirokayOptionsDocument],
-                **kwargs) -> DataFrame:
-    """Create a DataFrame from a file or an existing DataFrame and
-    apply Deirokay treatments to correctly parse it and pre-validate
-    its content.
-
-    Parameters
-    ----------
-    data : Union[str, DataFrame]
-        [description]
-    options : Union[dict, str]
-        Either a `dict` or a local/S3 path to an YAML/JSON file.
-
-    Returns
-    -------
-    DataFrame
-        A pandas DataFrame treated by Deirokay.
-    """
-    if isinstance(options, str):
-        options_dict = fs_factory(options).read_dict()
-    else:
-        options_dict = options
-
-    options_dict.update(kwargs)
-
-    columns = options_dict.pop('columns')
-
-    if isinstance(data, str):
-        df = pandas_read(data, columns=list(columns), **options_dict)
-    else:
-        df = data.copy()[list(columns)]
-    data_treater(df, columns)
-
-    return df
-
-
-def pandas_read(data: str, columns: List[str], sql: bool = False,
-                **kwargs) -> DataFrame:
-    """Infer the file type by its extension and call the proper
-    `pandas` method to parse it.
-
-    Parameters
-    ----------
-    data : str
-        Path to file or SQL query.
-    columns : List[str]
-        List of columns to be parsed.
-    sql : bool, optional
-        Whether or not `data` should be interpreted as a path to a file
-        or a SQL query.
-
-
-    Returns
-    -------
-    DataFrame
-        The pandas DataFrame.
-    """
-    default_kwargs: Dict[str, Any]
-    if sql:
-        default_kwargs = {
-            'columns': columns
-        }
-        default_kwargs.update(kwargs)
-        return read_sql(data, **default_kwargs)
-
-    file_extension = splitext(data)[1].lstrip('.')
-
-    if file_extension == 'sql':
-        default_kwargs = {
-            'columns': columns
-        }
-        default_kwargs.update(kwargs)
-        query = fs_factory(data).read()
-        return read_sql(query, **default_kwargs)
-
-    elif file_extension == 'csv':
-        default_kwargs = {
-            'dtype': str,
-            'skipinitialspace': True,
-            'usecols': columns
-        }
-        default_kwargs.update(kwargs)
-        return read_csv(data, **default_kwargs)
-
-    elif file_extension == 'parquet':
-        default_kwargs = {
-            'columns': columns
-        }
-        default_kwargs.update(kwargs)
-        return read_parquet(data, **default_kwargs)
-
-    elif file_extension in ('xls', 'xlsx'):
-        default_kwargs = {
-            'usecols': columns
-        }
-        return read_excel(data, **kwargs)
-    else:
-        read_ = getattr(pandas, f'read_{file_extension}', None)
-        if read_ is None:
-            raise TypeError(f'File type "{file_extension}" not supported')
-        return read_(data, **kwargs)[columns]
-
-
-def get_dtype_treater(dtype: Any) -> Type[treaters.Validator]:
-    """Map a dtype to its Treater class."""
-
-    treat_dtypes = {
-        DTypes.INT64: treaters.IntegerTreater,
-        DTypes.FLOAT64: treaters.FloatTreater,
-        DTypes.STRING: treaters.StringTreater,
-        DTypes.DATETIME: treaters.DateTime64Treater,
-        DTypes.DATE: treaters.DateTreater,
-        DTypes.TIME: treaters.TimeTreater,
-        DTypes.BOOLEAN: treaters.BooleanTreater,
-        DTypes.DECIMAL: treaters.DecimalTreater,
-    }
-    treat_primitives = {
-        int: treaters.IntegerTreater,
-        float: treaters.FloatTreater,
-        str: treaters.StringTreater,
-        Timestamp: treaters.DateTime64Treater,
-        datetime.date: treaters.DateTreater,
-        datetime.time: treaters.TimeTreater,
-        bool: treaters.BooleanTreater,
-        decimal.Decimal: treaters.DecimalTreater,
-    }
-
-    try:
-        if isinstance(dtype, DTypes):
-            return treat_dtypes[dtype]
-        elif isinstance(dtype, str):
-            return treat_dtypes[DTypes(dtype)]
-        else:
-            return treat_primitives[dtype]
-
-    except KeyError as e:
-        raise NotImplementedError(f"Handler for '{dtype}' hasn't been"
-                                  " implemented yet") from e
-
-
-def get_treater_instance(option: DeirokayOption) -> treaters.Validator:
-    """Create a treater instance from a Deirokay-style option.
-
-    Example
-    -------
-
-    .. code-block:: python
-
-        option = {
-            'dtype': 'integer',
-            'thousand_sep': ','
-        }
-    """
-    option = option.copy()
-    dtype = option.pop('dtype')
-
-    cls = get_dtype_treater(dtype)
-    return cls(**option)
-
-
-def data_treater(df: DataFrame, options: dict) -> None:
-    """Receive options dict and call the proper treater class for each
-    Deirokay data type.
-
-    Parameters
-    ----------
-    df : DataFrame
-        Raw DataFrame to be treated.
-    options : dict
-        Deirokay options.
-
-    Raises
-    ------
-    NotImplementedError
-        Data type not valid or not implemented.
-    """
-    _check_columns_in_df_columns(options.keys(), df.columns)
-
-    for col, opt in options.items():
-        option: dict = opt.copy()
-
-        dtype = option.get('dtype', None)
-        rename_to = option.pop('rename', None)
-
-        if dtype is not None:
-            try:
-                df[col] = get_treater_instance(option)(df[col])
-            except Exception as e:
-                raise Exception(f'Error when parsing "{col}".') from e
-
-        if rename_to is not None:
-            df.rename(columns={col: rename_to}, inplace=True)
diff --git a/deirokay/parser/reader/__init__.py b/deirokay/parser/reader/__init__.py
new file mode 100644
index 0000000..dd9b0cc
--- /dev/null
+++ b/deirokay/parser/reader/__init__.py
@@ -0,0 +1,48 @@
+"""
+Classes and methods to load data from several sources into Deirokay.
+"""
+
+from typing import List, Type
+
+from deirokay._typing import DeirokayReadCallable
+from deirokay.backend import MultiBackendMixin
+from deirokay.enums import Backend
+
+from .dask_reader import read as read_dask
+from .pandas_reader import read as read_pandas
+
+
+class DataReader(MultiBackendMixin):
+    """Helper class to load data from a data source in function of the
+    backend."""
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
+    read: DeirokayReadCallable
+
+
+DataReader.register_backend_method('read',
+                                   staticmethod(read_dask),  # type: ignore  # Staticmethod is Callable # noqa: E501
+                                   Backend.DASK)
+DataReader.register_backend_method('read',
+                                   staticmethod(read_pandas),  # type: ignore  # Staticmethod is Callable # noqa: E501
+                                   Backend.PANDAS)
+
+
+def reader_factory(backend: Backend) -> Type[DataReader]:
+    """Return the proper reader for the given backend.
+
+    Parameters
+    ----------
+    backend : Backend
+        Backend.
+
+    Returns
+    -------
+    Type[DataReader]
+        Subclass of DataReader containing a `read` staticmethod.
+    """
+    return DataReader.attach_backend(backend)
+
+
+__all__ = (
+    'reader_factory',
+)
diff --git a/deirokay/parser/reader/dask_reader.py b/deirokay/parser/reader/dask_reader.py
new file mode 100644
index 0000000..87f97ca
--- /dev/null
+++ b/deirokay/parser/reader/dask_reader.py
@@ -0,0 +1,75 @@
+from posixpath import splitext
+from typing import Any, Dict, List, Union
+
+import dask.dataframe  # lazy module
+
+
+def read(data: Union[str, 'dask.dataframe.DataFrame'],
+         columns: List[str],
+         sql: bool = False,
+         **kwargs) -> 'dask.dataframe.DataFrame':
+    """Infer the file type by its extension and call the proper
+    `dask.dataframe` method to parse it.
+
+    Parameters
+    ----------
+    data : Union[str, dask.dataframe.DataFrame]
+        Path to file or SQL query, or DataFrame object
+    columns : List[str]
+        List of columns to be parsed.
+    sql : bool, optional
+        Whether or not `data` should be interpreted as a path to a file
+        or a SQL query.
+    **kwargs : dict
+        Arguments to be passed to `dask.dataframe` methods when reading.
+
+    Returns
+    -------
+    dask.dataframe.DataFrame
+        The dask DataFrame.
+    """
+    default_kwargs: Dict[str, Any] = {}
+
+    if isinstance(data, dask.dataframe.DataFrame):
+        return data[columns].copy()
+
+    if not isinstance(data, str):
+        raise TypeError(f'Unexpected type for `data` ({data.__class__})')
+
+    if sql:
+        read_ = dask.dataframe.read_sql
+
+    else:
+        file_extension = splitext(data)[1].lstrip('.').lower()
+
+        if file_extension == 'csv':
+            default_kwargs.update({
+                'dtype': str,
+                'skipinitialspace': True,
+            })
+
+        read_ = getattr(dask.dataframe, f'read_{file_extension}', None)
+        if read_ is None:
+            raise TypeError(f'File type "{file_extension}" not supported')
+
+    default_kwargs.update(kwargs)
+
+    # try `columns` argument
+    try:
+        read_data = read_(data, columns=columns, **default_kwargs)
+    except TypeError as e:
+        if 'columns' not in str(e):
+            raise e
+    # try `usecols` argument
+    try:
+        read_data = read_(data, usecols=columns, **default_kwargs)
+    except TypeError as e:
+        if 'usecols' not in str(e):
+            raise e
+    # give up, read everything, filter columns later
+    read_data = read_(data, **default_kwargs)[columns]
+
+    if sql:
+        read_data = read_data.reset_index()
+
+    return read_data
diff --git a/deirokay/parser/reader/pandas_reader.py b/deirokay/parser/reader/pandas_reader.py
new file mode 100644
index 0000000..ccbeb7e
--- /dev/null
+++ b/deirokay/parser/reader/pandas_reader.py
@@ -0,0 +1,80 @@
+from posixpath import splitext
+from typing import Any, Dict, List, Union
+
+import pandas  # lazy module
+
+from deirokay.fs import fs_factory
+
+
+def read(data: Union[str, 'pandas.DataFrame'],
+         columns: List[str],
+         sql: bool = False,
+         **kwargs) -> 'pandas.DataFrame':
+    """Infer the file type by its extension and call the proper
+    `pandas` method to parse it.
+
+    Parameters
+    ----------
+    data : Union[DataFrame, dd.DataFrame, str]
+        Path to file or SQL query, or DataFrame object
+    columns : List[str]
+        List of columns to be parsed.
+    sql : bool, optional
+        Whether or not `data` should be interpreted as a path to a file
+        or a SQL query.
+    **kwargs : dict
+        Arguments to be passed to `pandas` methods when reading.
+
+
+    Returns
+    -------
+    pandas.DataFrame
+        The pandas DataFrame.
+    """
+    default_kwargs: Dict[str, Any] = {}
+
+    if isinstance(data, pandas.DataFrame):
+        return data[columns].copy()
+
+    if not isinstance(data, str):
+        raise TypeError(f'Unexpected type for `data` ({data.__class__})')
+
+    if sql:
+        read_ = pandas.read_sql
+
+    else:
+        file_extension = splitext(data)[1].lstrip('.').lower()
+
+        if file_extension == 'sql':
+            query = fs_factory(data).read()
+            return read(query, columns, sql=True, **kwargs)
+
+        if file_extension in ('xls', 'xlsx'):
+            file_extension = 'excel'
+
+        if file_extension == 'csv':
+            default_kwargs.update({
+                'dtype': str,
+                'skipinitialspace': True,
+            })
+
+        read_ = getattr(pandas, f'read_{file_extension}', None)
+        if read_ is None:
+            raise TypeError(f'File type "{file_extension}" not supported')
+
+    default_kwargs.update(kwargs)
+
+    # try `columns` argument
+    try:
+        return read_(data, columns=columns, **default_kwargs)
+    except TypeError as e:
+        if 'columns' not in str(e):
+            raise e
+    # try `usecols` argument
+    try:
+        return read_(data, usecols=columns, **default_kwargs)
+    except TypeError as e:
+        if 'usecols' not in str(e):
+            raise e
+    # give up, read everything, filter columns later
+    return read_(data, **default_kwargs)[columns]
diff --git a/deirokay/parser/treaters.py b/deirokay/parser/treaters.py
deleted file mode 100644
index f2b8bd8..0000000
--- a/deirokay/parser/treaters.py
+++ /dev/null
@@ -1,398 +0,0 @@
-"""
-Classes and functions to treat column data types according to
-Deirokay data types.
-"""
-
-from decimal import Decimal
-from typing import Iterable, List, Optional, Union
-
-from numpy import nan
-from pandas import NA, NaT, Series, to_datetime
-
-
-class Validator():
-    """Base validation class for column data type validation.
-    """
-
-    def __init__(self, *, unique: bool = False, nullable: bool = True):
-        self.unique = unique
-        self.nullable = nullable
-
-    def __call__(self, listlike: Iterable) -> Series:
-        return self.treat(Series(listlike))
-
-    def treat(self, series: Series) -> Series:
-        """Treat a raw Series to match data expectations for parsing
-        and formatting.
-
-        Parameters
-        ----------
-        series : Series
-            Raw pandas Series to be treated.
-
-        Raises
-        ------
-        ValueError
-            Column has null values when not_null constraint was
-            requested or
-            column has duplicate values when unique constraint was
-            requested.
-        """
-        if not self.nullable and any(series.isnull()):
-            null_indices = list(series[series.isnull()].index)
-            null_indices_limit = null_indices[:min(len(null_indices), 30)]
-            raise ValueError(f"The '{series.name}' column has"
-                             f" {len(null_indices)} null values,"
-                             " but it shouldn't.\n"
-                             "Here are the indices of some null values:\n"
-                             f"{null_indices_limit}...")
-
-        if self.unique and not series.is_unique:
-            duplicated = list(series[series.duplicated(keep='first')])
-            duplicated_limit = duplicated[:min(len(duplicated), 10)]
-
-            raise ValueError(f"The '{series.name}' column values"
-                             " are not unique, as requested.\n"
-                             f"There are {len(duplicated)} non unique values,"
-                             " and here are some of them:\n"
-                             f"{duplicated_limit}...")
-
-        return series
-
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        """Create a Deirokay-compatible serializable object that can
-        be serialized (in JSON or YAML formats) and parsed back by
-        Deirokay treaters.
-        This method is useful when generating validation documents
-        that embed Deirokay-compatible user data.
-        The output format is a Python dict containing two keys:
-        - `values`: list of values as Python object.
-        - `parser`: Deirokay options to parse the data.
-        (See more: `deirokay.parser.get_treater_instance`).
-
-        Parameters
-        ----------
-        series : Series
-            Pandas Series to be serialized.
-
-        Returns
-        -------
-        dict
-            A Python dict containing the keys `values` and `parser`.
-        """
-        raise NotImplementedError('No serializer for this treater.')
-
-
-class NumericTreater(Validator):
-    """Base class for numeric treaters"""
-
-    def __init__(self, thousand_sep: Optional[str] = None, **kwargs):
-        super().__init__(**kwargs)
-
-        self.thousand_sep = thousand_sep
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = super().treat(series)
-        series = self._treat_thousand_sep(series)
-
-        return series
-
-    def _treat_thousand_sep(self, series: Series) -> Series:
-        if self.thousand_sep is not None:
-            try:
-                series = series.str.replace(self.thousand_sep, '', regex=False)
-            except AttributeError as e:
-                raise AttributeError(
-                    'Make sure you are not declaring a `thousand_sep` to'
-                    ' read a non-text-like column. This may happen when'
-                    ' reading numeric columns from a .parquet file,'
-                    ' for instance.'
-                ) from e
-        return series
-
-
-class BooleanTreater(Validator):
-    """Treater for boolean-like variables"""
-
-    def __init__(self,
-                 truthies: List[str] = ['true', 'True'],
-                 falsies: List[str] = ['false', 'False'],
-                 ignore_case: bool = False,
-                 default_value: Optional[bool] = None,
-                 **kwargs):
-        super().__init__(**kwargs)
-
-        assert default_value in (True, False, None)
-
-        self.ignore_case = ignore_case
-        self.default_value = default_value
-
-        if self.ignore_case:
-            self.truthies = set(truthy.lower() for truthy in truthies)
-            self.falsies = set(falsy.lower() for falsy in falsies)
-        else:
-            self.truthies = set(truthies)
-            self.falsies = set(falsies)
-
-        if self.truthies & self.falsies:
-            raise ValueError('Truthies and Falsies sets should be'
-                             ' disjoint sets.')
-
-    def _evaluate(self, value: Union[bool, str, None]) -> Union[bool, None]:
-        if value is True:
-            return True
-        if value is False:
-            return False
-        if value is None or value is nan or value is NA:
-            return self.default_value
-
-        _value = value.lower() if self.ignore_case else value
-        if _value in self.truthies:
-            return True
-        if _value in self.falsies:
-            return False
-        raise ValueError(f'Unexpected boolean value: "{value}"'
-                         f' ({value.__class__})\n'
-                         f'Expected values: {self.truthies | self.falsies}')
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = super().treat(series)
-        series = series.apply(self._evaluate).astype('boolean')
-        # Validate again
-        super().treat(series)
-
-        return series
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is NA:
-                return None
-            return bool(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'boolean'
-            }
-        }
-
-
-class FloatTreater(NumericTreater):
-    """Treater for float variables"""
-
-    def __init__(self, decimal_sep: Optional[str] = None, **kwargs):
-        super().__init__(**kwargs)
-
-        self.decimal_sep = decimal_sep
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = super().treat(series)
-        series = self._treat_decimal_sep(series)
-
-        return series.astype(float).astype('Float64')
-
-    def _treat_decimal_sep(self, series: Series) -> Series:
-        if self.decimal_sep is not None and self.decimal_sep != '.':
-            try:
-                series = series.str.replace(self.decimal_sep, '.', regex=False)
-            except AttributeError as e:
-                print(*e.args)
-                raise AttributeError(
-                    'Make sure you are not declaring a `decimal_sep` to'
-                    ' read a non-text-like column. This may happen when'
-                    ' reading numeric columns from a .parquet file,'
-                    ' for instance.'
-                )
-        return series
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is NA:
-                return None
-            return float(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'float'
-            }
-        }
-
-
-class DecimalTreater(FloatTreater):
-    """Treater for decimal variables"""
-
-    def __init__(self, decimal_places: Optional[int] = None, **kwargs):
-        super().__init__(**kwargs)
-
-        self.decimal_places = decimal_places
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = NumericTreater.treat(self, series)
-        series = self._treat_decimal_sep(series)
-        series = series.map(lambda x: Decimal(x) if x is not None else None)
-        series = self._treat_decimal_places(series)
-
-        return series
-
-    def _treat_decimal_places(self, series: Series) -> Series:
-        if self.decimal_places is not None:
-            q = Decimal(10) ** -self.decimal_places
-            series = series.apply(
-                lambda x: x.quantize(q) if x is not None else None
-            )
-        return series
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is None or item is NA:
-                return None
-            return str(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'decimal'
-            }
-        }
-
-
-class IntegerTreater(NumericTreater):
-    """Treater for integer variables"""
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        return super().treat(series).astype(float).astype('Int64')
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is NA:
-                return None
-            return int(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'integer'
-            }
-        }
-
-
-class DateTime64Treater(Validator):
-    """Treater for datetime variables"""
-
-    def __init__(self, format: str = '%Y-%m-%d %H:%M:%S', **kwargs):
-        super().__init__(**kwargs)
-
-        self.format = format
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = super().treat(series)
-
-        return to_datetime(series, format=self.format)
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is None or item is NaT:
-                return None
-            return str(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'datetime'
-            }
-        }
-
-
-class DateTreater(DateTime64Treater):
-    """Treater for date-only variables"""
-
-    def __init__(self, format: str = '%Y-%m-%d', **kwargs):
-        super().__init__(format, **kwargs)
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        return super().treat(series).dt.date
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is None or item is NaT:
-                return None
-            return str(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'date'
-            }
-        }
-
-
-class TimeTreater(DateTime64Treater):
-    """Treater for time-only variables"""
-
-    def __init__(self, format: str = '%H:%M:%S', **kwargs):
-        super().__init__(format, **kwargs)
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        return super().treat(series).dt.time
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is None or item is NaT:
-                return None
-            return str(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'time'
-            }
-        }
-
-
-class StringTreater(Validator):
-    """Treater for string variables"""
-
-    def __init__(self, treat_null_as: Optional[str] = None, **kwargs):
-        super().__init__(**kwargs)
-
-        self.treat_null_as = treat_null_as
-
-    # docstr-coverage:inherited
-    def treat(self, series: Series) -> Series:
-        series = super().treat(series)
-
-        if self.treat_null_as is not None:
-            series = series.fillna(self.treat_null_as)
-
-        return series
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def serialize(series: Series) -> dict:
-        def _convert(item):
-            if item is None:
-                return None
-            return str(item)
-        return {
-            'values': [_convert(item) for item in series],
-            'parser': {
-                'dtype': 'string'
-            }
-        }
diff --git a/deirokay/parser/treaters/__init__.py b/deirokay/parser/treaters/__init__.py
new file mode 100644
index 0000000..7ecce66
--- /dev/null
+++ b/deirokay/parser/treaters/__init__.py
@@ -0,0 +1,60 @@
+"""
+Classes and methods to treat data immediately after being loaded
+into Deirokay.
+"""
+
+from typing import Any, Type
+
+from deirokay._typing import DeirokayOption
+from deirokay._utils import recursive_subclass_generator
+from deirokay.enums import Backend, DTypes
+
+from .builtin import BaseTreater
+
+DTYPE_2_TREATER = {
+    cls.supported_dtype: cls
+    for cls in recursive_subclass_generator(BaseTreater)
+    if cls.supported_dtype is not None
+}
+
+PRIMITIVE_2_TREATER = {
+    primitive: cls
+    for cls in recursive_subclass_generator(BaseTreater)
+    for primitive in cls.supported_primitives
+}
+
+
+def get_dtype_treater(dtype: Any) -> Type[BaseTreater]:
+    """Map a dtype to its Treater class."""
+    try:
+        if isinstance(dtype, DTypes):
+            return DTYPE_2_TREATER[dtype]
+        elif isinstance(dtype, str):
+            return DTYPE_2_TREATER[DTypes(dtype)]
+        else:
+            return PRIMITIVE_2_TREATER[dtype]
+
+    except KeyError as e:
+        raise NotImplementedError(f"Handler for '{dtype}' hasn't been"
+                                  " implemented yet") from e
+
+
+def get_treater_instance(option: DeirokayOption,
+                         backend: Backend) -> BaseTreater:
+    """Create a treater instance from a Deirokay-style option.
+
+    Example
+    -------
+
+    .. code-block:: python
+
+        option = {
+            'dtype': 'integer',
+            'thousand_sep': ','
+        }
+    """
+    option = option.copy()
+    dtype = option.pop('dtype')
+
+    cls = get_dtype_treater(dtype)
+    return cls.attach_backend(backend)(**option)
diff --git a/deirokay/parser/treaters/builtin/__init__.py b/deirokay/parser/treaters/builtin/__init__.py
new file mode 100644
index 0000000..6c2d104
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/__init__.py
@@ -0,0 +1,21 @@
+"""
+Implementation of builtin Deirokay treaters.
+"""
+from deirokay._utils import recursive_subclass_generator
+
+from .base_treater import BaseTreater
+from .boolean_treater import BooleanTreater  # noqa: F401
+from .date_treater import DateTreater  # noqa: F401
+from .datetime64_treater import DateTime64Treater  # noqa: F401
+from .decimal_treater import DecimalTreater  # noqa: F401
+from .float_treater import FloatTreater  # noqa: F401
+from .integer_treater import IntegerTreater  # noqa: F401
+from .numeric_treater import NumericTreater  # noqa: F401
+from .string_treater import StringTreater  # noqa: F401
+from .time_treater import TimeTreater  # noqa: F401
+from .validator import Validator  # noqa: F401
+
+__all__ = tuple(
+    cls.__name__
+    for cls in recursive_subclass_generator(BaseTreater)
+)
diff --git a/deirokay/parser/treaters/builtin/base_treater.py b/deirokay/parser/treaters/builtin/base_treater.py
new file mode 100644
index 0000000..6cc4e06
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/base_treater.py
@@ -0,0 +1,78 @@
+from abc import ABC
+from typing import Any, Iterable, List, Optional
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayDataSeries, DeirokaySerializedSeries
+from deirokay.backend import MultiBackendMixin
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import treat as treat_
+
+
+class BaseTreater(MultiBackendMixin, ABC):
+    """Base class for all data treaters."""
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
+    supported_dtype: Optional[DTypes] = None
+    """Optional[DTypes]: DType treated by this treater."""
+    supported_primitives: List[Any] = []
+    """List[Any]: List of primitives supported by this treater."""
+
+    def __call__(self, series: DeirokayDataSeries) -> DeirokayDataSeries:
+        """Proxy for `treat`."""
+        return self.treat(series)
+
+    def treat(self, series: Iterable) -> DeirokayDataSeries:
+        """Treat a raw Series to match data expectations for parsing
+        and formatting.
+
+        Calls the appropriate method for the detected backend
+
+        Parameters
+        ----------
+        series : Iterable
+            Raw series to be treated.
+        """
+        raise NotImplementedError
+
+    @treat_(Backend.PANDAS, force=True)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        if isinstance(series, pandas.Series):
+            return series
+        return pandas.Series(series)
+
+    @treat_(Backend.DASK, force=True)
+    def _treat_dask(self, series: Iterable) -> 'dask.dataframe.Series':
+        if isinstance(series, dask.dataframe.Series):
+            return series
+
+        # Assumption: When treating a general Iterable, we assume
+        # it fits entirely in memory
+        # TODO: let user propose a max partition size,
+        # instead of assuming a single partition
+        return dask.dataframe.from_pandas(pandas.Series(series), npartitions=1)
+
+    @staticmethod
+    def serialize(series: DeirokayDataSeries) -> DeirokaySerializedSeries:
+        """Create a Deirokay-compatible serializable object that can
+        be serialized (in JSON or YAML formats) and parsed back by
+        Deirokay treaters.
+        This method is useful when generating validation documents
+        that embed Deirokay-compatible user data.
+        The output format is a Python dict containing two keys:
+        - `values`: list of values as Python object.
+        - `parser`: Deirokay options to parse the data.
+        (See more: `deirokay.parser.get_treater_instance`).
+
+        Parameters
+        ----------
+        series : DeirokayDataSeries
+            Series data to be serialized.
+
+        Returns
+        -------
+        dict
+            A Python dict containing the keys `values` and `parser`.
+        """
+        raise NotImplementedError
diff --git a/deirokay/parser/treaters/builtin/boolean_treater.py b/deirokay/parser/treaters/builtin/boolean_treater.py
new file mode 100644
index 0000000..61535a9
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/boolean_treater.py
@@ -0,0 +1,129 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+from typing import Iterable, List, Optional, Union
+
+import dask.dataframe  # lazy module
+import numpy  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .validator import Validator
+
+
+class BooleanTreater(Validator):
+    """Treater for boolean-like variables
+
+    Parameters
+    ----------
+    truthies : List[str], optional
+        List of strings to be considered as True,
+        by default ['true', 'True']
+    falsies : List[str], optional
+        List of strings to be considered as False,
+        by default ['false', 'False']
+    ignore_case : bool, optional
+        Ignore case when evaluating list of values,
+        by default False
+    default_value : Optional[bool], optional
+        Value to replace null values (None, pandas.NA or numpy,NaN).
+        By default None.
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.BOOLEAN
+    supported_primitives = [bool]
+
+    def __init__(self,
+                 truthies: List[str] = ['true', 'True'],
+                 falsies: List[str] = ['false', 'False'],
+                 ignore_case: bool = False,
+                 default_value: Optional[bool] = None,
+                 **kwargs):
+        super().__init__(**kwargs)
+
+        assert default_value in (True, False, None)
+
+        self.ignore_case = ignore_case
+        self.default_value = default_value
+
+        if self.ignore_case:
+            self.truthies = set(truthy.lower() for truthy in truthies)
+            self.falsies = set(falsy.lower() for falsy in falsies)
+        else:
+            self.truthies = set(truthies)
+            self.falsies = set(falsies)
+
+        if self.truthies & self.falsies:
+            raise ValueError('Truthies and Falsies sets should be'
+                             ' disjoint sets.')
+
+    def _evaluate(self, value: Union[bool, str, None]) -> Union[bool, None]:
+        if value is True:
+            return True
+        if value is False:
+            return False
+        if value is None or value is numpy.nan or value is pandas.NA:
+            return self.default_value
+
+        _value = value.lower() if self.ignore_case else value
+        if _value in self.truthies:
+            return True
+        if _value in self.falsies:
+            return False
+        raise ValueError(f'Unexpected boolean value: "{value}"'
+                         f' ({value.__class__})\n'
+                         f'Expected values: {self.truthies | self.falsies}')
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = super()._treat_pandas(series)
+        series = series.apply(self._evaluate).astype('boolean')
+        # Validate again
+        super()._treat_pandas(series)
+
+        return series
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = super()._treat_dask(series)
+        series = series.apply(
+            self._evaluate, meta=(series.name, 'object')
+        ).astype('boolean')
+        # Validate again
+        super()._treat_dask(series)
+
+        return series
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is pandas.NA:
+                return None
+            return bool(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': BooleanTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(
+        series: 'pandas.Series'
+    ) -> DeirokaySerializedSeries:
+        return BooleanTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(
+        series: 'dask.dataframe.Series'
+    ) -> DeirokaySerializedSeries:
+        return BooleanTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/date_treater.py b/deirokay/parser/treaters/builtin/date_treater.py
new file mode 100644
index 0000000..1b520f4
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/date_treater.py
@@ -0,0 +1,68 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+import datetime
+from typing import Any, Iterable, List
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .datetime64_treater import DateTime64Treater
+
+
+class DateTreater(DateTime64Treater):
+    """Treater for date-only variables"
+
+    Parameters
+    ----------
+    format : str, optional
+        Format to parse dates from, by default '%Y-%m-%d'
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.DATE
+    supported_primitives: List[Any] = [datetime.date]
+
+    def __init__(self, format: str = '%Y-%m-%d', **kwargs):
+        super().__init__(format, **kwargs)
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        return super()._treat_pandas(series).dt.date
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        return super()._treat_dask(series).dt.date
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is None or item is pandas.NaT:
+                return None
+            return str(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': DateTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(
+        series: 'pandas.Series'
+    ) -> DeirokaySerializedSeries:
+        return DateTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(
+        series: 'dask.dataframe.Series'
+    ) -> DeirokaySerializedSeries:
+        return DateTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/datetime64_treater.py b/deirokay/parser/treaters/builtin/datetime64_treater.py
new file mode 100644
index 0000000..c7f12c6
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/datetime64_treater.py
@@ -0,0 +1,81 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+import datetime
+from typing import Iterable
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .validator import Validator
+
+_supported_primitives = [datetime.datetime]
+try:
+    _supported_primitives.append(pandas.Timestamp)
+except ModuleNotFoundError:
+    pass
+
+
+class DateTime64Treater(Validator):
+    """Treater for datetime variables
+
+    Parameters
+    ----------
+    format : str, optional
+        Format to parse dates from, by default '%Y-%m-%d %H:%M:%S'
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.DATETIME
+    supported_primitives = _supported_primitives
+
+    def __init__(self, format: str = '%Y-%m-%d %H:%M:%S', **kwargs):
+        super().__init__(**kwargs)
+
+        self.format = format
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = super()._treat_pandas(series)
+
+        return pandas.to_datetime(series, format=self.format)
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = super()._treat_dask(series)
+
+        return dask.dataframe.to_datetime(series, format=self.format)
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is None or item is pandas.NaT:
+                return None
+            return str(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': DateTime64Treater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(
+        series: 'pandas.Series'
+    ) -> DeirokaySerializedSeries:
+        return DateTime64Treater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(
+        series: 'dask.dataframe.Series'
+    ) -> DeirokaySerializedSeries:
+        return DateTime64Treater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/decimal_treater.py b/deirokay/parser/treaters/builtin/decimal_treater.py
new file mode 100644
index 0000000..03efeb8
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/decimal_treater.py
@@ -0,0 +1,98 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+import decimal
+from decimal import Decimal
+from typing import Any, Iterable, List, Optional
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayDataSeries, DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .float_treater import FloatTreater
+from .numeric_treater import NumericTreater
+
+
+class DecimalTreater(FloatTreater):
+    """Treater for decimal variables
+
+    Parameters
+    ----------
+    decimal_places : Optional[int], optional
+        Number of decimal places, by default None
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.DECIMAL
+    supported_primitives: List[Any] = [decimal.Decimal]
+
+    def __init__(self, decimal_places: Optional[int] = None, **kwargs):
+        super().__init__(**kwargs)
+
+        self.decimal_places = decimal_places
+
+    def _treat_decimal_places(
+        self, series: DeirokayDataSeries, **kwargs
+    ) -> DeirokayDataSeries:
+        if self.decimal_places is not None:
+            q = Decimal(10) ** -self.decimal_places
+            series = series.apply(
+                lambda x: x.quantize(q) if x is not None else None,
+                **kwargs
+            )
+        return series
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = NumericTreater._treat_pandas(self, series)
+        series = self._treat_decimal_sep(series)
+        series = series.map(lambda x: Decimal(x) if x is not None else None)
+        series = self._treat_decimal_places(series)
+
+        return series
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = NumericTreater._treat_dask(self, series)
+        series = self._treat_decimal_sep(series)
+        series = series.map(
+            lambda x: Decimal(x) if x is not None else None,
+            meta=(series.name, 'object')
+        )
+        series = self._treat_decimal_places(series,
+                                            meta=(series.name, 'object'))
+
+        return series
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is None or item is pandas.NA:
+                return None
+            return str(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': DecimalTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(
+        series: 'pandas.Series'
+    ) -> DeirokaySerializedSeries:
+        return DecimalTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(
+        series: 'dask.dataframe.Series'
+    ) -> DeirokaySerializedSeries:
+        return DecimalTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/float_treater.py b/deirokay/parser/treaters/builtin/float_treater.py
new file mode 100644
index 0000000..e1d59cb
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/float_treater.py
@@ -0,0 +1,89 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+from typing import Iterable, Optional
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayDataSeries, DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .numeric_treater import NumericTreater
+
+
+class FloatTreater(NumericTreater):
+    """Treater for float variables
+
+    Parameters
+    ----------
+    decimal_sep : Optional[str], optional
+        Character to use as decimal separator, by default None
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.FLOAT64
+    supported_primitives = [float]
+
+    def __init__(self, decimal_sep: Optional[str] = None, **kwargs):
+        super().__init__(**kwargs)
+
+        self.decimal_sep = decimal_sep
+
+    def _treat_decimal_sep(self,
+                           series: DeirokayDataSeries
+                           ) -> DeirokayDataSeries:
+        if self.decimal_sep is not None and self.decimal_sep != '.':
+            try:
+                series = series.str.replace(self.decimal_sep, '.', regex=False)
+            except AttributeError as e:
+                print(*e.args)
+                raise AttributeError(
+                    'Make sure you are not declaring a `decimal_sep` to'
+                    ' read a non-text-like column. This may happen when'
+                    ' reading numeric columns from a .parquet file,'
+                    ' for instance.'
+                )
+        return series
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = super()._treat_pandas(series)
+        series = self._treat_decimal_sep(series)
+
+        return series.astype(float).astype('Float64')
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = super()._treat_dask(series)
+        series = self._treat_decimal_sep(series)
+
+        return series.astype(float).astype('Float64')
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is pandas.NA:
+                return None
+            return float(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': FloatTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(series: 'pandas.Series') -> DeirokaySerializedSeries:
+        return FloatTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(series: 'dask.dataframe.Series'
+                        ) -> DeirokaySerializedSeries:
+        return FloatTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/integer_treater.py b/deirokay/parser/treaters/builtin/integer_treater.py
new file mode 100644
index 0000000..8f2e8e7
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/integer_treater.py
@@ -0,0 +1,57 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+from typing import Iterable
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .numeric_treater import NumericTreater
+
+
+class IntegerTreater(NumericTreater):
+    """Treater for integer variables"""
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.INT64
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        return super()._treat_pandas(series).astype(float).astype('Int64')
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        return super()._treat_dask(series).astype(float).astype('Int64')
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is pandas.NA:
+                return None
+            return int(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': IntegerTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(
+        series: 'pandas.Series'
+    ) -> DeirokaySerializedSeries:
+        return IntegerTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(
+        series: 'dask.dataframe.Series'
+    ) -> DeirokaySerializedSeries:
+        return IntegerTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/numeric_treater.py b/deirokay/parser/treaters/builtin/numeric_treater.py
new file mode 100644
index 0000000..6dc4672
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/numeric_treater.py
@@ -0,0 +1,61 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+from typing import Iterable, Optional
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayDataSeries
+from deirokay.enums import Backend
+
+from ..multibackend import treat
+from .validator import Validator
+
+
+class NumericTreater(Validator):
+    """Base class for numeric treaters
+
+    Parameters
+    ----------
+    thousand_sep : Optional[str], optional
+        Character to use as thousand separator, by default None
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+
+    def __init__(self, thousand_sep: Optional[str] = None, **kwargs):
+        super().__init__(**kwargs)
+
+        self.thousand_sep = thousand_sep
+
+    def _treat_thousand_sep(self,
+                            series: DeirokayDataSeries) -> DeirokayDataSeries:
+        if self.thousand_sep is not None:
+            try:
+                series = series.str.replace(self.thousand_sep, '', regex=False)
+            except AttributeError as e:
+                raise AttributeError(
+                    'Make sure you are not declaring a `thousand_sep` to'
+                    ' read a non-text-like column. This may happen when'
+                    ' reading numeric columns from a .parquet file,'
+                    ' for instance.'
+                ) from e
+        return series
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = super()._treat_pandas(series)
+        series = self._treat_thousand_sep(series)
+
+        return series
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = super()._treat_dask(series)
+        series = self._treat_thousand_sep(series)
+
+        return series
diff --git a/deirokay/parser/treaters/builtin/string_treater.py b/deirokay/parser/treaters/builtin/string_treater.py
new file mode 100644
index 0000000..5883843
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/string_treater.py
@@ -0,0 +1,78 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+
+from typing import Iterable, Optional
+
+import dask.dataframe  # lazy module
+import numpy  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .validator import Validator
+
+
+class StringTreater(Validator):
+    """Treater for string variables
+
+    Parameters
+    ----------
+    treat_null_as : Optional[str], optional
+        Character to replace null values for, by default None
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.STRING
+    supported_primitives = [str]
+
+    def __init__(self, treat_null_as: Optional[str] = None, **kwargs):
+        super().__init__(**kwargs)
+
+        self.treat_null_as = treat_null_as
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        series = super()._treat_pandas(series)
+
+        if self.treat_null_as is not None:
+            series = series.fillna(self.treat_null_as)
+
+        return series
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        series = super()._treat_dask(series)
+
+        if self.treat_null_as is not None:
+            series = series.fillna(self.treat_null_as)
+
+        return series
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is None or item is pandas.NA or item is numpy.NaN:
+                return None
+            return str(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': StringTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(series: 'pandas.Series') -> DeirokaySerializedSeries:
+        return StringTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(series: 'dask.dataframe.Series'
+                        ) -> DeirokaySerializedSeries:
+        return StringTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/time_treater.py b/deirokay/parser/treaters/builtin/time_treater.py
new file mode 100644
index 0000000..308a4f4
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/time_treater.py
@@ -0,0 +1,65 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+import datetime
+from typing import Any, Iterable, List
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokaySerializedSeries
+from deirokay.enums import Backend, DTypes
+
+from ..multibackend import serialize, treat
+from .datetime64_treater import DateTime64Treater
+
+
+class TimeTreater(DateTime64Treater):
+    """Treater for time-only variables
+
+    Parameters
+    ----------
+    format : str, optional
+        Format to parse times from, by default '%H:%M:%S'
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+    supported_dtype = DTypes.TIME
+    supported_primitives: List[Any] = [datetime.time]
+
+    def __init__(self, format: str = '%H:%M:%S', **kwargs):
+        super().__init__(format, **kwargs)
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        return super()._treat_pandas(series).dt.time
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        return super()._treat_dask(series).dt.time
+
+    @staticmethod
+    def _serialize_common(series):
+        def _convert(item):
+            if item is None or item is pandas.NaT:
+                return None
+            return str(item)
+        return {
+            'values': [_convert(item) for item in series],
+            'parser': {
+                'dtype': TimeTreater.supported_dtype.value
+            }
+        }
+
+    @serialize(Backend.PANDAS)
+    @staticmethod
+    def _serialize_pandas(series: 'pandas.Series') -> DeirokaySerializedSeries:
+        return TimeTreater._serialize_common(series)
+
+    @serialize(Backend.DASK)
+    @staticmethod
+    def _serialize_dask(series: 'dask.dataframe.Series'
+                        ) -> DeirokaySerializedSeries:
+        return TimeTreater._serialize_common(series)
diff --git a/deirokay/parser/treaters/builtin/validator.py b/deirokay/parser/treaters/builtin/validator.py
new file mode 100644
index 0000000..c3db122
--- /dev/null
+++ b/deirokay/parser/treaters/builtin/validator.py
@@ -0,0 +1,132 @@
+"""
+Classes and functions to treat column data types according to
+Deirokay data types.
+"""
+from typing import Iterable
+
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay.enums import Backend
+
+from ..multibackend import treat
+from .base_treater import BaseTreater
+
+
+class Validator(BaseTreater):
+    """Base validation class for column data type validation.
+
+    Parameters
+    ----------
+    unique : bool, optional
+        Make sure values are unique, by default False
+    nullable : bool, optional
+        Allow values to be null, by default True
+    """
+    supported_backends = [Backend.PANDAS, Backend.DASK]
+
+    DISPLAY_NULL_INDICES_LIMIT = 30
+    DISPLAY_DUPL_INDICES_LIMIT = 10
+
+    def __init__(self, *,
+                 unique: bool = False,
+                 nullable: bool = True):
+        self.unique = unique
+        self.nullable = nullable
+
+    @treat(Backend.PANDAS)
+    def _treat_pandas(self, series: Iterable) -> 'pandas.Series':
+        """Treat a raw Series to match data expectations for parsing
+        and formatting.
+
+        Parameters
+        ----------
+        series : Iterable
+            Raw pandas Series to be treated.
+
+        Raises
+        ------
+        ValueError
+            Column has null values when not_null constraint was
+            requested or
+            column has duplicate values when unique constraint was
+            requested.
+        """
+        series = super()._treat_pandas(series)
+
+        if not self.nullable and any(series.isnull()):
+            null_indices = list(series[series.isnull()].index)
+            null_indices_limit = null_indices[:min(
+                len(null_indices), self.DISPLAY_NULL_INDICES_LIMIT
+            )]
+            raise ValueError(
+                f"The '{series.name}' column has {len(null_indices)} null"
+                " values, but it shouldn't.\n"
+                "Here are the indices of some null values:\n"
+                f"{null_indices_limit}..."
+            )
+
+        if self.unique and not series.is_unique:
+            duplicated = list(series[series.duplicated(keep='first')])
+            duplicated_limit = duplicated[:min(
+                len(duplicated), self.DISPLAY_DUPL_INDICES_LIMIT
+            )]
+            raise ValueError(
+                f"The '{series.name}' column values are not unique, as"
+                " requested.\n"
+                f"There are {len(duplicated)} non unique values, and here are"
+                " some of them:\n"
+                f"{duplicated_limit}..."
+            )
+
+        return series
+
+    @treat(Backend.DASK)
+    def _treat_dask(
+        self, series: Iterable
+    ) -> 'dask.dataframe.Series':
+        """Treat a raw Series to match data expectations for parsing
+        and formatting.
+
+        Parameters
+        ----------
+        series : Iterable
+            Raw dask Series to be treated.
+
+        Raises
+        ------
+        ValueError
+            Column has null values when not_null constraint was
+            requested or
+            column has duplicate values when unique constraint was
+            requested.
+        """
+        series = super()._treat_dask(series)
+
+        if not self.nullable and any(series.isnull()):
+            null_indices = series[series.isnull()].index
+            null_indices_display_limit = list(null_indices[:min(
+                len(null_indices), self.DISPLAY_NULL_INDICES_LIMIT
+            )])
+            raise ValueError(
+                f"The '{series.name}' column has {len(null_indices)} null"
+                " values, but it shouldn't.\n"
+                "Here are the indices of some null values:\n"
+                f"{null_indices_display_limit}..."
+            )
+
+        duplicated_bool = (series.value_counts() > 1)
+        if self.unique and any(duplicated_bool):
+            duplicated_values = duplicated_bool[duplicated_bool].index
+            duplicated_display_limit = list(duplicated_values[:min(
+                len(duplicated_values), self.DISPLAY_DUPL_INDICES_LIMIT
+            )])
+            raise ValueError(
+                f"The '{series.name}' column values are not unique, as"
+                " requested.\n"
+                f"There are {len(duplicated_values)} non unique values, and"
+                " here are some of them:\n"
+                f"{duplicated_display_limit}..."
+            )
+
+        return series
diff --git a/deirokay/parser/treaters/multibackend.py b/deirokay/parser/treaters/multibackend.py
new file mode 100644
index 0000000..389e516
--- /dev/null
+++ b/deirokay/parser/treaters/multibackend.py
@@ -0,0 +1,43 @@
+from deirokay._typing import GeneralDecorator
+from deirokay.backend import register_backend_method
+from deirokay.enums import Backend
+
+
+def treat(backend: Backend, *, force: bool = False) -> GeneralDecorator:
+    """Define a decorator that turns any method into an alias for the
+    `treat` method when executed using a given `backend`.
+
+    Parameters
+    ----------
+    backend : Backend
+        The backend for the decorated method.
+
+    Returns
+    -------
+    GeneralDecorator
+        A decorator for the alias method of `treat`.
+    force : bool, optional
+        Force overwrite target method when it already exists.
+        Defaults to False.
+    """
+    return register_backend_method('treat', backend, force=force)
+
+
+def serialize(backend: Backend, *, force: bool = False) -> GeneralDecorator:
+    """Define a decorator that turns any method into an alias for the
+    `serialize` method when executed using a given `backend`.
+
+    Parameters
+    ----------
+    backend : Backend
+        The backend for the decorated method.
+
+    Returns
+    -------
+    GeneralDecorator
+        A decorator for the alias method of `serialize`.
+    force : bool, optional
+        Force overwrite target method when it already exists.
+        Defaults to False.
+    """
+    return register_backend_method('serialize', backend, force=force)
diff --git a/deirokay/profiler.py b/deirokay/profiler.py
index 2a1cafd..0a2d2da 100644
--- a/deirokay/profiler.py
+++ b/deirokay/profiler.py
@@ -6,50 +6,55 @@
 import warnings
 from typing import List, Optional
 
-from pandas import DataFrame
+from deirokay.backend import detect_backend
+from deirokay.enums import Backend
+from deirokay.exceptions import UnsupportedBackend
 
 from .__version__ import __version__
-from ._typing import (DeirokayStatement, DeirokayValidationDocument,
-                      DeirokayValidationItem)
+from ._typing import (DeirokayDataSource, DeirokayStatement,
+                      DeirokayValidationDocument, DeirokayValidationItem)
 from .fs import fs_factory
 from .statements import STATEMENTS_MAP
 
 
-def _generate_statements(df_scope: DataFrame,
-                         table_only: bool) -> List[DeirokayStatement]:
+def _generate_statements(df_scope: DeirokayDataSource,
+                         backend: Backend) -> List[DeirokayStatement]:
 
     statements: List[DeirokayStatement] = []
 
     for stmt_cls in STATEMENTS_MAP.values():
-        if table_only and stmt_cls.table_only or not stmt_cls.table_only:
-            try:
-                statement = stmt_cls.profile(df_scope)
-                statements.append(statement)
-            except NotImplementedError:
-                pass
-            except Exception as e:
-                columns = list(df_scope.columns)
-                warnings.warn(
-                    f'Unexpected error when profiling scope {columns}'
-                    f' using {stmt_cls.__name__} statement: {e}\n\n'
-                    'Please, consider reporting this issue to the '
-                    'developers.',
-                    RuntimeWarning
-                )
+        try:
+            execution_cls = stmt_cls.attach_backend(backend)
+            statement = execution_cls.profile(df_scope)
+            statements.append(statement)
+        except UnsupportedBackend:
+            pass
+        except NotImplementedError:
+            pass
+        except Exception as e:
+            columns = list(df_scope.columns)
+            warnings.warn(
+                f'Unexpected error when profiling scope {columns}'
+                f' using {stmt_cls.__name__} statement: {e}\n\n'
+                'Please, consider reporting this issue to the '
+                'developers.',
+                RuntimeWarning
+            )
     return statements
 
 
-def _generate_items(df: DataFrame) -> List[DeirokayValidationItem]:
+def _generate_items(df: DeirokayDataSource,
+                    backend: Backend) -> List[DeirokayValidationItem]:
     items: List[DeirokayValidationItem] = []
 
     df_columns = list(df.columns)
-    scope__table_stmt = zip([df_columns] + df_columns,
-                            [True] + [False]*len(df_columns))
-    for scope, table_only in scope__table_stmt:
+    scope__table_stmt = [df_columns] + df_columns
+
+    for scope in scope__table_stmt:
         df_scope = df[scope] if isinstance(scope, list) else df[[scope]]
         item = {
             'scope': scope,
-            'statements': _generate_statements(df_scope, table_only=table_only)
+            'statements': _generate_statements(df_scope, backend)
         }  # type: DeirokayValidationItem
 
         if item['statements']:
@@ -58,7 +63,7 @@ def _generate_items(df: DataFrame) -> List[DeirokayValidationItem]:
     return items
 
 
-def profile(df: DataFrame,
+def profile(df: DeirokayDataSource,
             document_name: str,
             save_to: Optional[str] = None) -> DeirokayValidationDocument:
     """Generate a validation document from a given template DataFrame
@@ -89,10 +94,11 @@ def profile(df: DataFrame,
     dict
         The auto-generated validation document as Python `dict`.
     """
+    backend = detect_backend(df)
     validation_document = {
         'name': document_name,
         'description': f'Auto generated using Deirokay {__version__}',
-        'items': _generate_items(df)
+        'items': _generate_items(df, backend)
     }  # type: DeirokayValidationDocument
 
     if save_to:
diff --git a/deirokay/statements/__init__.py b/deirokay/statements/__init__.py
index 6318849..1e3e120 100644
--- a/deirokay/statements/__init__.py
+++ b/deirokay/statements/__init__.py
@@ -1,22 +1,20 @@
 """
-Module for BaseStatement and builtin Deirokay statements.
+Classes and methods to build and load builtin or custom Deirokay
+statements.
 """
+from typing import Dict, Type
 
-import inspect
+from deirokay._utils import recursive_subclass_generator
 
-from .base_statement import BaseStatement
-from .column_expression import ColumnExpression  # noqa F401
-from .contain import Contain  # noqa F401
-from .not_null import NotNull  # noqa F401
-from .row_count import RowCount  # noqa F401
-from .unique import Unique  # noqa F401
+from .builtin import BaseStatement
 
-STATEMENTS_MAP = {
+STATEMENTS_MAP: Dict[str, Type[BaseStatement]] = {
     cls.name: cls
-    for _, cls in locals().items()
-    if inspect.isclass(cls) and issubclass(cls, BaseStatement)
+    for cls in recursive_subclass_generator(BaseStatement)
+    if cls is not BaseStatement
 }
 
 __all__ = tuple(
-    cls.__name__ for cls in STATEMENTS_MAP.values()
+    cls.__name__
+    for cls in recursive_subclass_generator(BaseStatement)
 )
diff --git a/deirokay/statements/builtin/__init__.py b/deirokay/statements/builtin/__init__.py
new file mode 100644
index 0000000..d1091fc
--- /dev/null
+++ b/deirokay/statements/builtin/__init__.py
@@ -0,0 +1,16 @@
+"""
+Implementation of builtin Deirokay statements.
+"""
+from deirokay._utils import recursive_subclass_generator
+
+from .base_statement import BaseStatement
+from .column_expression import ColumnExpression  # noqa: F401
+from .contain import Contain  # noqa: F401
+from .not_null import NotNull  # noqa: F401
+from .row_count import RowCount  # noqa: F401
+from .unique import Unique  # noqa: F401
+
+__all__ = tuple(
+    cls.__name__
+    for cls in recursive_subclass_generator(BaseStatement)
+)
diff --git a/deirokay/statements/base_statement.py b/deirokay/statements/builtin/base_statement.py
similarity index 76%
rename from deirokay/statements/base_statement.py
rename to deirokay/statements/builtin/base_statement.py
index add8fe0..8de2dbc 100644
--- a/deirokay/statements/base_statement.py
+++ b/deirokay/statements/builtin/base_statement.py
@@ -2,13 +2,14 @@
 The base statement that all other statements inherit from.
 """
 from abc import ABC, abstractmethod
+from typing import List
 
-from pandas import DataFrame
+from deirokay._typing import (DeirokayDataSource, DeirokayOption,
+                              DeirokayStatement)
+from deirokay.backend import MultiBackendMixin
 
-from .._typing import DeirokayStatement
 
-
-class BaseStatement(ABC):
+class BaseStatement(MultiBackendMixin, ABC):
     """Base abstract statement class for all Deirokay statements.
 
     Attributes
@@ -17,23 +18,26 @@ class BaseStatement(ABC):
         Statement parameters provided by user.
     """
 
-    name = 'base_statement'
+    name: str = 'base_statement'
     """str: Statement name when referred in Validation Documents
     (only valid for Deirokay built-in statements)."""
-    expected_parameters = ['type', 'severity', 'location']
+    expected_parameters: List[str] = ['type', 'severity', 'location']
     """List[str]: Parameters expected for this statement."""
-    table_only = False
-    """bool: Whether or not this statement in applicable only to the
-    entire table, instead of scoped columns."""
 
-    def __init__(self, options: dict) -> None:
+    def __init_subclass__(cls) -> None:
+        """Validate subclassed statement."""
+        assert cls.name != BaseStatement.name, (
+            'You should specify a `name` attribute for your statement class.'
+        )
+
+    def __init__(self, options: DeirokayOption) -> None:
         self._validate_options(options)
         self.options = options
 
-    def _validate_options(self, options: dict) -> None:
+    @classmethod
+    def _validate_options(cls, options: DeirokayOption) -> None:
         """Make sure all provided statement parameters are expected
         by statement classes"""
-        cls = type(self)
         unexpected_parameters = [
             option for option in options
             if option not in (cls.expected_parameters +
@@ -46,7 +50,7 @@ def _validate_options(self, options: dict) -> None:
                 f'The valid parameters are: {cls.expected_parameters}'
             )
 
-    def __call__(self, df: DataFrame) -> dict:
+    def __call__(self, df: DeirokayDataSource) -> dict:
         """Run statement instance."""
         internal_report = self.report(df)
         result = self.result(internal_report)
@@ -57,8 +61,7 @@ def __call__(self, df: DataFrame) -> dict:
         }
         return final_report
 
-    @abstractmethod
-    def report(self, df: DataFrame) -> dict:
+    def report(self, df: DeirokayDataSource) -> dict:
         """Receive a DataFrame containing only columns on the scope of
         validation and returns a report of related metrics that can
         be used later to declare this Statement as fulfilled or
@@ -75,6 +78,7 @@ def report(self, df: DataFrame) -> dict:
         dict
             A dictionary of useful statistics about the target columns.
         """
+        raise NotImplementedError
 
     @abstractmethod
     def result(self, report: dict) -> bool:
@@ -95,7 +99,7 @@ def result(self, report: dict) -> bool:
         """
 
     @staticmethod
-    def profile(df: DataFrame) -> DeirokayStatement:
+    def profile(df: DeirokayDataSource) -> DeirokayStatement:
         """Given a template data table, generate a statement dict
         from it.
 
diff --git a/deirokay/statements/builtin/column_expression.py b/deirokay/statements/builtin/column_expression.py
new file mode 100644
index 0000000..348485a
--- /dev/null
+++ b/deirokay/statements/builtin/column_expression.py
@@ -0,0 +1,226 @@
+"""
+Statement to evaluate mathematical expressions againt a set of
+columns from a given scope.
+"""
+import re
+from decimal import Decimal
+from functools import reduce
+from typing import List, Tuple
+
+import dask  # lazy module
+import dask.dataframe  # lazy module
+import numpy  # lazy module
+import pandas  # lazy module
+
+from deirokay.enums import Backend
+
+from ..multibackend import report
+from .base_statement import BaseStatement
+
+
+class ColumnExpression(BaseStatement):
+    """
+    Evaluates an expression (or a list of expressions) involving
+    the scope columns, using `numpy.eval()`.
+    The statement passes only if all expressions evaluate to `True`.
+
+    The columns in the scope must be ideally of the same dtype.
+    This statement supports the following dtypes:
+    `string`, `integer`, `float` and `decimal`.
+
+    The available parameters for this statement are:
+
+    * `expressions` (required): an expression (or a `list` of
+      expressions) to be evaluated.
+      The valid operators are: `==`, `!=`, `=~`, `>=`, `<=`, `>` and
+      `<`.
+    * `at_least_%`: the minimum percentage of valid rows. Default: 100.
+    * `at_most_%`: the maximum percentage of valid rows. Default: 100.
+    * `rtol`: the relative tolerance for float evaluations (when
+      using the `=~` operator). Default: 1e-5.
+    * `atol`: the absolute tolerance for float evaluations (when
+      using the `=~` operator). Default: 1e-8.
+
+    Examples
+    --------
+    In the example below, in JSON format, we test whether or not the
+    values of the `a` column are equal to the values of the `b` column.
+    Similarly, we test whether or not the values of the `b` column are
+    greater than the values of the `c` column:
+
+    .. code-block:: json
+
+        {
+            "scope": ["a", "b", "c"],
+            "statements": [
+                {
+                    "type": "column_expression",
+                    "expressions": ["a == b", "a < c"],
+                    "at_least_%": 50.0
+                }
+            ]
+        }
+
+    For float comparisons, you may prefer using the `rtol` or `atol`
+    parameters, in addition to the `=~` operator. For example, if you
+    want to test whether or not the values of the `a` column are equal
+    to the values of the `b` column with a relative tolerance of 1e-3,
+    you can use the following JSON:
+
+    .. code-block:: json
+
+        {
+            "scope": ["a", "b"],
+            "statements": [
+                {
+                    "type": "column_expression",
+                    "expressions": "a =~ b",
+                    "rtol": 1e-3
+                }
+            ]
+        }
+
+    """
+
+    name = 'column_expression'
+    expected_parameters = [
+        'expressions',
+        'at_least_%',
+        'at_most_%',
+        'rtol',
+        'atol'
+    ]
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
+
+    VALID_OPERATORS = '|'.join(['==', '!=', '=~', '>=', '<=', '>', '<'])
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+        self.expressions = self.options['expressions']
+        self.at_least_perc = self.options.get('at_least_%', 100.0)
+        self.at_most_perc = self.options.get('at_most_%', 100.0)
+        self.rtol = self.options.get('rtol', 1e-5)
+        self.atol = self.options.get('atol', 1e-8)
+
+        if not isinstance(self.expressions, list):
+            self.expressions = [self.expressions]
+
+    def _convert_df_dtypes(self, df: 'pandas.DataFrame') -> 'pandas.DataFrame':
+        """
+        Fixes DataFrame dtypes. If Int64Dtype() or Float64Dtype(),
+        converts to traditional int64 and float64 dtypes. If object
+        dtype, we apply a more accurate check on the column type,
+        verifying the dtype of all its cells.
+
+        When a pandas version corrects this bug, we can delete this
+        method.
+        """
+        pandas_dtypes_int = [pandas.Int64Dtype(), pandas.Int32Dtype()]
+        pandas_dtypes_float = [pandas.Float64Dtype(), pandas.Float32Dtype()]
+        pandas_dtypes_decimal = [Decimal]
+
+        def _fix_column(column: pandas.Series):
+            if column.dtype in pandas_dtypes_int:
+                column = column.astype(int)
+            elif column.dtype in pandas_dtypes_float:
+                column = column.astype(float)
+            elif column.dtype == object:
+                if len(column.map(type).drop_duplicates()) > 1:
+                    raise Exception('Mixed types')
+                if type(column[0]) in pandas_dtypes_decimal:
+                    column = column.astype(float)
+            return column
+
+        return df.apply(_fix_column)
+
+    def _eval(self, df: 'pandas.DataFrame', expr: str) -> 'pandas.Series':
+        """
+        Accomplishes the paper of `pandas.eval` when we have the
+        `=~` comparison to evaluate. That implementation is done by
+        using `numpy.eval`.
+        """
+        if '=~' not in expr:
+            return df.eval(expr)
+
+        expr_terms = re.split(ColumnExpression.VALID_OPERATORS, expr)
+        expr_operators = re.findall(ColumnExpression.VALID_OPERATORS, expr)
+
+        if len(expr_terms) != len(expr_operators) + 1:
+            raise SyntaxError(
+                'Invalid expression. Incoherent number of expressions and'
+                ' comparison operators'
+            )
+
+        def _eval_part(term_1, operator, term_2):
+            if operator == '=~':
+                return pandas.Series(
+                    numpy.isclose(
+                        df.eval(term_1),
+                        df.eval(term_2),
+                        atol=self.atol,
+                        rtol=self.rtol
+                    )
+                )
+            return df.eval(term_1 + operator + term_2)
+
+        return reduce(
+            lambda x, y: x & y,
+            (
+                _eval_part(term_1, operator, term_2)
+                for term_1, operator, term_2
+                in zip(expr_terms, expr_operators, expr_terms[1:])
+            )
+        )
+
+    def _generate_report(self,
+                         summary: List[Tuple[int, int]],
+                         nrows: int) -> dict:
+        expressions_report = [
+            {
+                'expression': expr,
+                'valid_rows': valid,
+                'valid_rows_%': 100.0*valid/nrows,
+                'invalid_rows': invalid,
+                'invalid_rows_%': 100.0*invalid/nrows
+            }
+            for expr, (valid, invalid) in zip(self.expressions, summary)
+        ]
+        return {
+            'column_expressions': expressions_report
+        }
+
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df: 'pandas.DataFrame') -> dict:
+        df = self._convert_df_dtypes(df)
+
+        results = (
+            self._eval(df, expr)
+            for expr in self.expressions
+        )
+        summary = ((sum(result), sum(~result)) for result in results)
+        return self._generate_report(summary, len(df))
+
+    @report(Backend.DASK)
+    def _report_dask(self, df: 'dask.dataframe.DataFrame') -> dict:
+        df = df.map_partitions(self._convert_df_dtypes,
+                               meta=dict(df.dtypes.iteritems()))
+
+        results = (
+            dask.dataframe.from_delayed(
+                dask.delayed(self._eval)(partition, expr)
+                for partition in df.to_delayed()
+            )
+            for expr in self.expressions
+        )
+        summary = ((sum(result), sum(~result)) for result in results)
+        return self._generate_report(summary, len(df))
+
+    # docstr-coverage:inherited
+    def result(self, report: dict) -> bool:
+        for item in report['column_expressions']:
+            if not item['valid_rows_%'] >= self.at_least_perc:
+                return False
+            if not item['valid_rows_%'] <= self.at_most_perc:
+                return False
+        return True
diff --git a/deirokay/statements/builtin/contain.py b/deirokay/statements/builtin/contain.py
new file mode 100644
index 0000000..c21db84
--- /dev/null
+++ b/deirokay/statements/builtin/contain.py
@@ -0,0 +1,510 @@
+"""
+Statement to check the presence (or absence) of values in a scope.
+"""
+import warnings
+from typing import List
+
+import dask.dataframe  # lazy module
+import numpy  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayStatement
+from deirokay._utils import noneor
+from deirokay.enums import Backend
+from deirokay.parser import get_dtype_treater, get_treater_instance
+from deirokay.parser.loader import data_treater
+
+from ..multibackend import profile, report
+from .base_statement import BaseStatement
+
+NODEFAULT = object()
+
+
+class Contain(BaseStatement):
+    """
+    Checks if the given scope contains specific values. You may also
+    check the number of their occurrences by specifying a minimum and
+    maximum value of frequency.
+
+    The available parameters for this statement are:
+
+    * `rule` (required): One of `all`, `only` or `all_and_only`.
+    * `values` (required): A list of values to which the rule applies.
+    * `multicolumn`: A boolean indicating whether the statement should
+      consider each of the `values` as a tuple of multiple columns or
+      a single value. When set to False and evaluated over a scope
+      containing more than one column, the rule will be applied over
+      all the values from the original columns as in a single column.
+      Default: False.
+    * `parser`: The parser (or a list) to be used to parse the `values`.
+      Correspond to the `parser` parameter of the `treater` function
+      (see `deirokay.data_reader` method).
+      Either `parser` or `parsers` must be declared.
+    * `parsers`: An alias for `parser`, recommended when `multicolumn`
+      is set to True.
+      Either `parser` or `parsers` must be declared.
+    * `min_occurrences`: a global minimum number of occurrences for
+      each of the `values`. Default: 1 for `all` and `all_and_only`
+      rules, 0 for `only`.
+    * `max_occurrences`: a global maximum number of occurrences for
+      each of the `values`. Default: `inf` (unbounded).
+    * `occurrences_per_value`: a list of dictionaries overriding the
+      global boundaries. Each dictionary may have the following keys:
+
+        * `values` (required): a list of values to which the
+          occurrence bounds below must apply to.
+        * `min_occurrences`: a minimum number of occurrences for these
+          values. Default: global `min_occurrences` parameter.
+        * `max_occurrences`: a maximum number of occurrences for these
+          values. Default: global `max_occurrences` parameter.
+
+      Global parameters apply to all values not present in any of the
+      dictionaries in `occurrences_per_value` (but yet present on the
+      main `values` list).
+
+    * `report_limit`: if set to a positive integer, limit the number of
+      items generated in the statement report.
+      Default: 32.
+
+    The `all` rule checks if all the `values` declared are present in
+    the scope (possibly tolerating other values not declared in
+    `values`).
+    Use it when you want to be sure that your data contains at least
+    all the values you declare, also setting `min_occurrences` and
+    `max_occurrences` when necessary.
+    You may also check for "zero" occurrences of a set of values by
+    setting `max_occurrences` to 0.
+
+    The `only` rule ensures that the `values` are the only possible
+    values in the scope (possibly not containing them at all).
+    Use it when you want to enumerate the admitted values for the
+    scope, as in an enumeration.
+
+    The `all_and_only` rule checks both if all the `values` declared
+    are present in the scope and if only they are present (not
+    tolerating values not declared).
+    Use it when you know all the possible values for the scope and you
+    are sure that they will be always present.
+
+    The `min_occurrences` and `max_occurrences` parameters are applied
+    applied to all the `values` declared, and only these. It means you
+    cannot (yet) specify boundaries for values you did't declare.
+
+    Null values are considered valid for the purpose of the statement
+    evaluation and must be explicitely passed in `values` if you wish
+    to allow them (or not).
+
+    You may also notice that, by tweaking the expected number of
+    occurrences, you may end up having the very same behaviour
+    regardless the `rule` you choose.
+    In this case, you should go for the rule that semantically matches
+    best your intents, so that your final validation document looks
+    more readable and easy to understand.
+
+    Examples
+    --------
+    * You have a table of users containg a column `handedness`
+      only admitting the values:
+      `right-handed`, `left-handed` and `ambidextrous`.
+      You know that some of these values may not appear in the data,
+      but you don't want other values to be present.
+
+    .. code-block:: json
+
+        {
+            "scope": "handedness",
+            "statements": [
+                {
+                    "type": "contain",
+                    "rule": "only",
+                    "values": ["right-handed", "left-handed", "ambidextrous"],
+                    "parser": {"dtype": "string"}
+                }
+            ]
+        }
+
+    * You have a table of servers containg a column `role` which may
+      contain the values `master` and `slave`.
+      You want to be sure that there is always one and only one master
+      server in the data.
+
+    .. code-block:: json
+
+        {
+            "scope": "role",
+            "statements": [
+                {
+                    "type": "contain",
+                    "rule": "all",
+                    "values": ["master"],
+                    "parser": {"dtype": "string"},
+                    "min_occurrences": 1,
+                    "max_occurrences": 1
+                }
+            ]
+        }
+
+    You may also extend the previous example by making some adjustments
+    to ensure that there is no other value than `master` and `role`
+    in the data. Make notice that although the `rule` below is changed
+    to `only`, the statement above is still contemplated by the
+    `occurrences_per_value` parameter in the following validation item:
+
+    .. code-block:: json
+
+        {
+            "scope": "role",
+            "statements": [
+                {
+                    "type": "contain",
+                    "rule": "only",
+                    "values": ["master", "slave"],
+                    "parser": {"dtype": "string"},
+                    "occurrences_per_value": [
+                        {
+                            "values": ["master"],
+                            "min_occurrences": 1,
+                            "max_occurrences": 1
+                        }
+                    ]
+                }
+            ]
+        }
+
+    * You have a table of transactions containing details about
+      transactions in all the branches of a company. You expect that
+      there should always be at least one transaction per branch.
+
+    .. code-block:: json
+
+        {
+            "scope": ["branch_name"],
+            "statements": [
+                {
+                    "type": "contain",
+                    "rule": "all_and_only",
+                    "values": [
+                        "Albany", "Utica", "Scranton", "Akron",
+                        "Nashua", "Buffalo", "Rochester"
+                    ],
+                    "parser": {"dtype": "string"}
+                }
+            ]
+        }
+
+    * You have a table for the logs of user accesses to a website which
+      contains an `IP` column. You want to be sure that blacklisted
+      IPs are not present in the data. The following validation item in
+      YAML format checks for the absense of blacklisted IPs:
+
+    .. code-block:: yaml
+
+        scope: IP
+        statements:
+        - type: contain
+          rule: all
+          max_occurrences: 0
+          values: # blacklisted IPs
+          - 3.48.48.135
+          - 3.48.48.136
+          parser: {dtype: string}
+
+    * You want the pair 'San Diego' and '2022' to appear at most twice
+      in your dataset (for any reason).
+
+    .. code-block:: yaml
+
+        scope: [city, year]
+        statements:
+        - type: contain
+          rule: all
+          min_occurrences: 0
+          max_occurrences: 2
+          values:
+          - ['San Diego', 2022]
+          parsers:
+          - {dtype: string}
+          - {dtype: integer}
+
+    """
+    name = 'contain'
+    expected_parameters = [
+        'rule',
+        'values',
+        'multicolumn',
+        'parser',
+        'parsers',
+        'min_occurrences',
+        'max_occurrences',
+        'occurrences_per_value',
+        'report_limit',
+    ]
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
+
+    DEFAULT_MIN_OCCURRENCES = {
+        'all': (1, 0),
+        'only': (0, 0),
+        'all_and_only': (1, 0)
+    }
+    DEFAULT_MAX_OCCURRENCES = {
+        'all': (numpy.inf, numpy.inf),
+        'only': (numpy.inf, 0),
+        'all_and_only': (numpy.inf, 0)
+    }
+    DEFAULT_REPORT_LIMIT = 32
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+        self.rule = self.options['rule']
+        assert self.rule in ('all', 'only', 'all_and_only')
+        self.multicolumn = self.options.get('multicolumn', False)
+        _parsers = self.options.get('parser') or self.options['parsers']
+        if self.multicolumn:
+            self.parsers = _parsers
+        else:
+            self.parsers = [_parsers]
+        self.treaters = [
+            get_treater_instance(parser, backend=self.get_backend())
+            for parser in self.parsers
+        ]
+
+        self.min_occurrences = self.options.get('min_occurrences', None)
+        self.max_occurrences = self.options.get('max_occurrences', None)
+        self.occurrences_per_value = self.options.get(
+            'occurrences_per_value', []
+        )
+        self.report_limit = self.options.get('report_limit', NODEFAULT)
+
+        self._set_default_minmax_occurrences()
+
+    def _set_default_minmax_occurrences(self) -> None:
+        final_min = noneor(self.min_occurrences,
+                           Contain.DEFAULT_MIN_OCCURRENCES[self.rule][0])
+        final_max = noneor(self.max_occurrences,
+                           Contain.DEFAULT_MAX_OCCURRENCES[self.rule][0])
+        if (
+            self.max_occurrences is not None and
+            self.max_occurrences < final_min
+        ):
+            final_min = self.max_occurrences
+        if (
+            self.min_occurrences is not None and
+            self.min_occurrences > final_max
+        ):
+            final_max = self.min_occurrences
+        assert final_min >= 0
+        assert final_max >= 0
+        self.min_occurrences = final_min
+        self.max_occurrences = final_max
+
+    def _generate_analysis(self, value_counts):
+        if self.multicolumn:
+            def _unpack_row(row, *args):
+                return (*row, *args)
+        else:
+            def _unpack_row(row, *args):
+                return (row, *args)
+
+        listed_values = [
+            _unpack_row(
+                row,
+                self.min_occurrences,
+                self.max_occurrences
+            )
+            for row in self.options['values']
+        ]
+        occurences_per_value = [
+            _unpack_row(
+                row,
+                noneor(item.get('min_occurrences'), self.min_occurrences),
+                noneor(item.get('max_occurrences'), self.max_occurrences),
+            )
+            for item in self.occurrences_per_value
+            for row in item['values']
+        ]
+
+        occurrence_limits = pandas.DataFrame(
+            occurences_per_value + listed_values,
+            columns=value_counts.index.names + ['min', 'max']
+        )
+        options = {
+            col: parser
+            for col, parser in zip(value_counts.index.names, self.parsers)
+        }
+        data_treater(occurrence_limits, options, backend=Backend.PANDAS)
+
+        occurrence_limits.drop_duplicates(
+            subset=value_counts.index.names,
+            keep='first',
+            inplace=True
+        )
+        occurrence_limits.set_index(value_counts.index.names, inplace=True)
+
+        analysis = value_counts.to_frame().reset_index().merge(
+            occurrence_limits.reset_index(), how='outer'
+        )
+        analysis['count'].fillna(0, inplace=True)
+        analysis['min'].fillna(Contain.DEFAULT_MIN_OCCURRENCES[self.rule][1],
+                               inplace=True)
+        analysis['max'].fillna(Contain.DEFAULT_MAX_OCCURRENCES[self.rule][1],
+                               inplace=True)
+        analysis['result'] = (
+            analysis['count'].ge(analysis['min'])
+            &
+            analysis['count'].le(analysis['max'])
+        )
+        return analysis
+
+    def _generate_report(self, analysis):
+        columns = [analysis[col] for col in analysis.columns[:-4]]
+        serialized = (
+            treater.serialize(column)
+            for treater, column in zip(self.treaters, columns)
+        )
+        rows = zip(*(s['values'] for s in serialized))
+        values_report = sorted([
+            {
+                'value': value_row,
+                'count': analysis_row.count,
+                'result': analysis_row.result,
+            }
+            for value_row, analysis_row in zip(rows, analysis.itertuples())
+        ], key=lambda x: x['result'])
+
+        if (
+            self.report_limit is NODEFAULT and
+            len(values_report) > Contain.DEFAULT_REPORT_LIMIT
+        ):
+            self.report_limit = Contain.DEFAULT_REPORT_LIMIT
+            warnings.warn(
+                "The 'contain' statement's report size was automatically"
+                f' truncated to {Contain.DEFAULT_REPORT_LIMIT} items to'
+                '  prevent unexpectedly long logs.\n'
+                'If you wish to set a different'
+                ' size limit or even not set a limit at all (None),'
+                ' please declare the `report_limit` parameter explicitely.',
+                Warning
+            )
+
+        return {
+            'values': (
+                values_report if self.report_limit is NODEFAULT else
+                values_report if self.report_limit is None else
+                values_report[:self.report_limit]
+            )
+        }
+
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df: 'pandas.DataFrame') -> dict:
+        # Concat all columns
+        _cols = df.columns.tolist()
+
+        if not self.multicolumn:
+            # Columns are assumed to be of same Dtype
+            df = pandas.concat([df[col] for col in _cols]).to_frame()
+
+        value_counts = (
+            df.groupby(_cols, dropna=False)[_cols[0]].size()
+            .rename('count')
+        )
+        analysis = self._generate_analysis(value_counts)
+        return self._generate_report(analysis)
+
+    @report(Backend.DASK)
+    def _report_dask(self, df: 'dask.dataframe.DataFrame') -> dict:
+        # Concat all columns
+        _cols = df.columns.tolist()
+
+        if not self.multicolumn:
+            # Columns are assumed to be of same Dtype
+            df = dask.dataframe.concat([df[col] for col in _cols]).to_frame()
+
+        value_counts = (
+            df.groupby(_cols, dropna=False)[_cols[0]].size()
+            .rename('count')
+        )
+        analysis = self._generate_analysis(value_counts.compute())
+        return self._generate_report(analysis)
+
+    # docstr-coverage:inherited
+    def result(self, report: dict) -> bool:
+        return all(
+            item['result'] for item in report['values']
+        )
+
+    @profile(Backend.PANDAS)
+    @staticmethod
+    def _profile_pandas(df: 'pandas.DataFrame') -> DeirokayStatement:
+        if any(dtype != df.dtypes for dtype in df.dtypes):
+            raise NotImplementedError(
+                "Refusing to mix up different types of columns"
+            )
+
+        series = pandas.concat(df[col] for col in df.columns)
+
+        unique_series = series.drop_duplicates().dropna()
+        if len(unique_series) > 20:
+            raise NotImplementedError("Won't generate too long statements!")
+
+        value_frequency = series.value_counts()
+        min_occurrences = int(value_frequency.min())
+
+        statement_template = {
+            'type': 'contain',
+            'rule': 'all'
+        }  # type: DeirokayStatement
+        # Get most common type to infer treater
+        try:
+            statement_template.update(
+                get_dtype_treater(unique_series.map(type).mode()[0])
+                .attach_backend(Backend.PANDAS)
+                .serialize(unique_series)  # type: ignore
+            )
+        except TypeError:
+            raise NotImplementedError("Can't handle mixed types")
+        # Sort allowing `None` values, which will appear last
+        statement_template['values'].sort(key=lambda x: (x is None, x))
+
+        if min_occurrences != 1:
+            statement_template['min_occurrences'] = min_occurrences
+
+        return statement_template
+
+    @profile(Backend.DASK)
+    @staticmethod
+    def _profile_dask(df: 'dask.dataframe.DataFrame') -> DeirokayStatement:
+        if any(dtype != df.dtypes for dtype in df.dtypes):
+            raise NotImplementedError(
+                "Refusing to mix up different types of columns"
+            )
+
+        series = dask.dataframe.concat([df[col] for col in df.columns])
+
+        unique_series = series.drop_duplicates().dropna()
+        if len(unique_series) > 20:
+            raise NotImplementedError("Won't generate too long statements!")
+
+        value_frequency = series.value_counts()
+        min_occurrences = int(value_frequency.min().compute())
+
+        statement_template = {
+            'type': 'contain',
+            'rule': 'all'
+        }  # type: DeirokayStatement
+        # Get most common type to infer treater
+        try:
+            statement_template.update(
+                get_dtype_treater(unique_series.map(type).mode().compute()[0])
+                .attach_backend(Backend.DASK)
+                .serialize(unique_series)  # type: ignore
+            )
+        except TypeError:
+            raise NotImplementedError("Can't handle mixed types")
+        # Sort allowing `None` values, which will appear last
+        statement_template['values'].sort(key=lambda x: (x is None, x))
+
+        if min_occurrences != 1:
+            statement_template['min_occurrences'] = min_occurrences
+
+        return statement_template
diff --git a/deirokay/statements/not_null.py b/deirokay/statements/builtin/not_null.py
similarity index 75%
rename from deirokay/statements/not_null.py
rename to deirokay/statements/builtin/not_null.py
index f52ae25..8123432 100644
--- a/deirokay/statements/not_null.py
+++ b/deirokay/statements/builtin/not_null.py
@@ -1,9 +1,13 @@
 """
 Statement to check the number of not-null rows in a scope.
 """
-from pandas import DataFrame
+import dask.dataframe  # lazy module
+import pandas  # lazy module
 
-from .._typing import DeirokayStatement
+from deirokay._typing import DeirokayStatement
+from deirokay.enums import Backend
+
+from ..multibackend import profile, report
 from .base_statement import BaseStatement
 
 
@@ -92,6 +96,7 @@ class NotNull(BaseStatement):
 
     name = 'not_null'
     expected_parameters = ['at_least_%', 'at_most_%', 'multicolumn_logic']
+    supported_backends = [Backend.PANDAS, Backend.DASK]
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
@@ -102,21 +107,29 @@ def __init__(self, *args, **kwargs) -> None:
 
         assert self.multicolumn_logic in ('any', 'all')
 
-    # docstr-coverage:inherited
-    def report(self, df: DataFrame) -> dict:
+    def _report_common(self, df):
         if self.multicolumn_logic == 'all':
             #  REMINDER: ~all == any
             not_nulls = ~df.isnull().any(axis=1)
         else:
             not_nulls = ~df.isnull().all(axis=1)
 
-        report = {
-            'null_rows': int((~not_nulls).sum()),
-            'null_rows_%': float(100.0*(~not_nulls).sum()/len(not_nulls)),
-            'not_null_rows': int(not_nulls.sum()),
-            'not_null_rows_%': float(100.0*not_nulls.sum()/len(not_nulls)),
+        null_rows = int(sum(~not_nulls))
+        not_null_rows = int(sum(not_nulls))
+        return {
+            'null_rows': null_rows,
+            'null_rows_%': 100.0*null_rows/len(df),
+            'not_null_rows': not_null_rows,
+            'not_null_rows_%': 100.0*not_null_rows/len(df),
         }
-        return report
+
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df: 'pandas.DataFrame') -> dict:
+        return self._report_common(df)
+
+    @report(Backend.DASK)
+    def _report_dask(self, df: 'dask.dataframe.DataFrame') -> dict:
+        return self._report_common(df)
 
     # docstr-coverage:inherited
     def result(self, report: dict) -> bool:
@@ -126,16 +139,14 @@ def result(self, report: dict) -> bool:
             return False
         return True
 
-    # docstr-coverage:inherited
     @staticmethod
-    def profile(df: DataFrame) -> DeirokayStatement:
-        not_nulls = ~df.isnull().all(axis=1)
-
+    def _profile_common(df):
         statement = {
             'type': 'not_null'
         }  # type: DeirokayStatement
 
-        at_least_perc = float(100.0*not_nulls.sum()/len(not_nulls))
+        not_nulls = ~df.isnull().all(axis=1)
+        at_least_perc = float(100.0*sum(not_nulls)/len(not_nulls))
 
         if at_least_perc == 0.0:
             raise NotImplementedError(
@@ -146,3 +157,13 @@ def profile(df: DataFrame) -> DeirokayStatement:
             statement['at_least_%'] = at_least_perc
 
         return statement
+
+    @profile(Backend.PANDAS)
+    @staticmethod
+    def _profile_pandas(df: 'pandas.DataFrame') -> DeirokayStatement:
+        return NotNull._profile_common(df)
+
+    @profile(Backend.DASK)
+    @staticmethod
+    def _profile_dask(df: 'dask.dataframe.DataFrame') -> DeirokayStatement:
+        return NotNull._profile_common(df)
diff --git a/deirokay/statements/row_count.py b/deirokay/statements/builtin/row_count.py
similarity index 84%
rename from deirokay/statements/row_count.py
rename to deirokay/statements/builtin/row_count.py
index 8c5e019..ef0435b 100644
--- a/deirokay/statements/row_count.py
+++ b/deirokay/statements/builtin/row_count.py
@@ -1,9 +1,15 @@
 """
 Statement to check the number of rows in a scope.
 """
-from pandas import DataFrame
+from typing import List
 
-from .._typing import DeirokayStatement
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayStatement
+from deirokay.enums import Backend
+
+from ..multibackend import profile, report
 from .base_statement import BaseStatement
 
 
@@ -112,6 +118,7 @@ class RowCount(BaseStatement):
 
     name = 'row_count'
     expected_parameters = ['min', 'max', 'distinct']
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
@@ -120,16 +127,22 @@ def __init__(self, *args, **kwargs) -> None:
         self.max = self.options.get('max', None)
         self.distinct = self.options.get('distinct', False)
 
-    # docstr-coverage:inherited
-    def report(self, df: DataFrame) -> dict:
+    def _report_common(self, df):
         row_count = len(df)
         distinct_count = len(df.drop_duplicates())
 
-        report = {
+        return {
             'rows': row_count,
             'distinct_rows': distinct_count,
         }
-        return report
+
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df: 'pandas.DataFrame') -> dict:
+        return self._report_common(df)
+
+    @report(Backend.DASK)
+    def _report_dask(self, df: 'dask.dataframe.DataFrame') -> dict:
+        return self._report_common(df)
 
     # docstr-coverage:inherited
     def result(self, report: dict) -> bool:
@@ -146,9 +159,8 @@ def result(self, report: dict) -> bool:
                 return False
         return True
 
-    # docstr-coverage:inherited
     @staticmethod
-    def profile(df: DataFrame) -> DeirokayStatement:
+    def _profile_common(df):
         statement: DeirokayStatement
 
         if len(df.columns) > 1:
@@ -167,3 +179,13 @@ def profile(df: DataFrame) -> DeirokayStatement:
                 'max': count
             }
         return statement
+
+    @profile(Backend.PANDAS)
+    @staticmethod
+    def _profile_pandas(df: 'pandas.DataFrame') -> DeirokayStatement:
+        return RowCount._profile_common(df)
+
+    @profile(Backend.DASK)
+    @staticmethod
+    def _profile_dask(df: 'dask.dataframe.DataFrame') -> DeirokayStatement:
+        return RowCount._profile_common(df)
diff --git a/deirokay/statements/unique.py b/deirokay/statements/builtin/unique.py
similarity index 53%
rename from deirokay/statements/unique.py
rename to deirokay/statements/builtin/unique.py
index f1b2dc6..5f59353 100644
--- a/deirokay/statements/unique.py
+++ b/deirokay/statements/builtin/unique.py
@@ -1,9 +1,15 @@
 """
 Statement to check the number unique rows in a scope.
 """
-from pandas import DataFrame
+from typing import List
 
-from .._typing import DeirokayStatement
+import dask.dataframe  # lazy module
+import pandas  # lazy module
+
+from deirokay._typing import DeirokayStatement
+from deirokay.enums import Backend
+
+from ..multibackend import profile, report
 from .base_statement import BaseStatement
 
 
@@ -38,36 +44,47 @@ class Unique(BaseStatement):
 
     name = 'unique'
     expected_parameters = ['at_least_%']
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
         self.at_least_perc = self.options.get('at_least_%', 100.0)
 
-    # docstr-coverage:inherited
-    def report(self, df: DataFrame) -> dict:
-        unique = ~df.duplicated(keep=False)
-
-        report = {
-            'unique_rows': int(unique.sum()),
-            'unique_rows_%': float(100.0*unique.sum()/len(unique)),
+    @staticmethod
+    def _unique_rows(df):
+        """Get number of unique rows in DataFrame"""
+        _cols = df.columns.tolist()
+        value_counts = df.groupby(_cols, dropna=False)[_cols[0]].size()
+        return int(sum(value_counts == 1))
+
+    def _report_common(self, df):
+        unique_rows = Unique._unique_rows(df)
+        return {
+            'unique_rows': unique_rows,
+            'unique_rows_%': 100.0*unique_rows/len(df),
         }
-        return report
+
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df: 'pandas.DataFrame') -> dict:
+        return self._report_common(df)
+
+    @report(Backend.DASK)
+    def _report_dask(self, df: 'dask.dataframe.DataFrame') -> dict:
+        return self._report_common(df)
 
     # docstr-coverage:inherited
     def result(self, report: dict) -> bool:
         return report.get('unique_rows_%') >= self.at_least_perc
 
-    # docstr-coverage:inherited
     @staticmethod
-    def profile(df: DataFrame) -> DeirokayStatement:
-        unique = ~df.duplicated(keep=False)
-
+    def _profile_common(df):
         statement = {
             'type': 'unique',
         }  # type: DeirokayStatement
 
-        at_least_perc = float(100.0*unique.sum()/len(unique))
+        unique_rows = Unique._unique_rows(df)
+        at_least_perc = 100.0*unique_rows/len(df)
 
         if at_least_perc == 0.0:
             raise NotImplementedError(
@@ -78,3 +95,13 @@ def profile(df: DataFrame) -> DeirokayStatement:
             statement['at_least_%'] = at_least_perc
 
         return statement
+
+    @profile(Backend.PANDAS)
+    @staticmethod
+    def _profile_pandas(df: 'pandas.DataFrame') -> DeirokayStatement:
+        return Unique._profile_common(df)
+
+    @profile(Backend.DASK)
+    @staticmethod
+    def _profile_dask(df: 'dask.dataframe.DataFrame') -> DeirokayStatement:
+        return Unique._profile_common(df)
diff --git a/deirokay/statements/column_expression.py b/deirokay/statements/column_expression.py
deleted file mode 100644
index ed78ee5..0000000
--- a/deirokay/statements/column_expression.py
+++ /dev/null
@@ -1,201 +0,0 @@
-"""
-Statement to evaluate mathematical expressions againt a set of
-columns from a given scope.
-"""
-import re
-from decimal import Decimal
-
-import numpy
-from pandas import (DataFrame, Float32Dtype, Float64Dtype, Int32Dtype,
-                    Int64Dtype, Series)
-
-from .base_statement import BaseStatement
-
-
-class ColumnExpression(BaseStatement):
-    """
-    Evaluates an expression (or a list of expressions) involving
-    the scope columns, using `numpy.eval()`.
-    The statement passes only if all expressions evaluate to `True`.
-
-    The columns in the scope must be ideally of the same dtype.
-    This statement supports the following dtypes:
-    `string`, `integer`, `float` and `decimal`.
-
-    The available parameters for this statement are:
-
-    * `expressions` (required): an expression (or a `list` of
-      expressions) to be evaluated.
-      The valid operators are: `==`, `!=`, `=~`, `>=`, `<=`, `>` and
-      `<`.
-    * `at_least_%`: the minimum percentage of valid rows. Default: 100.
-    * `at_most_%`: the maximum percentage of valid rows. Default: 100.
-    * `rtol`: the relative tolerance for float evaluations (when
-      using the `=~` operator). Default: 1e-5.
-    * `atol`: the absolute tolerance for float evaluations (when
-      using the `=~` operator). Default: 1e-8.
-
-    Examples
-    --------
-    In the example below, in JSON format, we test whether or not the
-    values of the `a` column are equal to the values of the `b` column.
-    Similarly, we test whether or not the values of the `b` column are
-    greater than the values of the `c` column:
-
-    .. code-block:: json
-
-        {
-            "scope": ["a", "b", "c"],
-            "statements": [
-                {
-                    "type": "column_expression",
-                    "expressions": ["a == b", "a < c"],
-                    "at_least_%": 50.0
-                }
-            ]
-        }
-
-    For float comparisons, you may prefer using the `rtol` or `atol`
-    parameters, in addition to the `=~` operator. For example, if you
-    want to test whether or not the values of the `a` column are equal
-    to the values of the `b` column with a relative tolerance of 1e-3,
-    you can use the following JSON:
-
-    .. code-block:: json
-
-        {
-            "scope": ["a", "b"],
-            "statements": [
-                {
-                    "type": "column_expression",
-                    "expressions": "a =~ b",
-                    "rtol": 1e-3
-                }
-            ]
-        }
-
-    """
-
-    name = 'column_expression'
-    expected_parameters = [
-        'expressions', 'at_least_%', 'at_most_%', 'rtol', 'atol'
-    ]
-    table_only = False
-
-    def __init__(self, *args, **kwargs) -> None:
-        super().__init__(*args, **kwargs)
-
-        self.expressions = self.options['expressions']
-        self.at_least_perc = self.options.get('at_least_%', 100.0)
-        self.at_most_perc = self.options.get('at_most_%', 100.0)
-        self.rtol = self.options.get('rtol', 1e-5)
-        self.atol = self.options.get('atol', 1e-8)
-
-        if isinstance(self.expressions, str):
-            self.expressions = [self.expressions]
-
-        self.valid_expressions = ['==', '!=', '=~', '>=', '<=', '>', '<']
-        for expr in self.expressions:
-            if not any([x in expr for x in self.valid_expressions]):
-                raise SyntaxError('Expression comparison operand not found')
-
-    # docstr-coverage:inherited
-    def report(self, df: DataFrame) -> dict:
-        report = {}
-        df = df.copy()
-        df = self._fix_df_dtypes(df)
-        for expr in self.expressions:
-            row_count = self._eval(df, expr)
-            report[expr] = {
-                'valid_rows': int((row_count).sum()),
-                'valid_rows_%': float(
-                    100.0*(row_count).sum()/len(row_count)
-                ),
-                'invalid_rows': int((~row_count).sum()),
-                'invalid_rows_%': float(
-                    100.0*(~row_count).sum()/len(row_count)
-                )
-            }
-        return report
-
-    def _fix_df_dtypes(self, df: DataFrame) -> DataFrame:
-        """
-        Fixes DataFrame dtypes. If Int64Dtype() or Float64Dtype(),
-        converts to traditional int64 and float64 dtypes. If object
-        dtype, we apply a more accurate check on the column type,
-        verifying the dtype of all its cells.
-
-        When a pandas version corrects this bug, we can delete this
-        method.
-        """
-        pandas_dtypes_int = [Int64Dtype(), Int32Dtype()]
-        pandas_dtypes_float = [Float64Dtype(), Float32Dtype()]
-        pandas_dtypes_decimal = [Decimal]
-        for col in df.columns:
-            if df[[col]].dtypes.isin(pandas_dtypes_int)[0]:
-                df[[col]] = df[[col]].astype(int)
-            elif df[[col]].dtypes.isin(pandas_dtypes_float)[0]:
-                df[[col]] = df[[col]].astype(float)
-            elif df[[col]].dtypes[0] == object:
-                col_dtype = list(
-                    dict.fromkeys(
-                        [type(x) for x in df[col].to_numpy()]
-                    )
-                )
-                if len(col_dtype) > 1:
-                    raise Exception('Mixed types in column')
-                if col_dtype[0] in pandas_dtypes_decimal:
-                    df[[col]] = df[[col]].astype(float)
-        return df
-
-    def _eval(self, df: DataFrame, expr: str) -> Series:
-        if '=~' not in expr:
-            row_count = df.eval(expr)
-        else:
-            row_count = self._isclose_eval(df, expr)
-        return row_count
-
-    def _isclose_eval(self, df: DataFrame, expr: str) -> Series:
-        """
-        Accomplishes the paper of `pandas.eval` when we have the
-        `=~` comparison to evaluate. That implementation is done by
-        using `numpy.eval`.
-        """
-        expr_calculus = re.split('|'.join(self.valid_expressions), expr)
-        expr_comparison = re.findall('|'.join(self.valid_expressions), expr)
-        expr_comparison.append(None)
-
-        if len(expr_calculus) != len(expr_comparison):
-            raise SyntaxError(
-                """Invalid expression. Incoherent number of expressions and
-                comparison operands"""
-            )
-
-        eval_bool = Series(len(df) * [True])
-
-        for i in range(len(expr_calculus) - 1):
-            comparison_to_eval = (
-                expr_calculus[i] + expr_comparison[i] + expr_calculus[i+1]
-            )
-
-            if expr_comparison[i] != '=~':
-                eval_bool = df.eval(comparison_to_eval) & eval_bool
-            else:
-                eval_bool = Series(
-                    numpy.isclose(
-                        df.eval(expr_calculus[i]),
-                        df.eval(expr_calculus[i+1]),
-                        atol=self.atol,
-                        rtol=self.rtol
-                    )
-                ) & eval_bool
-        return eval_bool
-
-    # docstr-coverage:inherited
-    def result(self, report: dict) -> bool:
-        for expr in report.keys():
-            if not report[expr].get('valid_rows_%') >= self.at_least_perc:
-                return False
-            if not report[expr].get('valid_rows_%') <= self.at_most_perc:
-                return False
-        return True
diff --git a/deirokay/statements/contain.py b/deirokay/statements/contain.py
deleted file mode 100644
index 665c749..0000000
--- a/deirokay/statements/contain.py
+++ /dev/null
@@ -1,438 +0,0 @@
-"""
-Statement to check the presence (or absence) of values in a scope.
-"""
-from numpy import inf
-from pandas import DataFrame, concat
-
-from .._typing import DeirokayStatement
-from ..parser import get_dtype_treater, get_treater_instance
-from .base_statement import BaseStatement
-
-
-class Contain(BaseStatement):
-    """
-    Checks if the given scope contains specific values. We can also
-    check the number of their occurrences by specifying a minimum and
-    maximum value of frequency.
-
-    The available parameters for this statement are:
-
-    * `rule` (required): One of `all`, `only` or `all_and_only`.
-    * `values` (required): A list of values to which the rule applies.
-    * `parser` (required): The parser to be used to parse the `values`.
-      Correspond to the `parser` parameter of the `treater` function
-      (see `deirokay.data_reader` method).
-    * `min_occurrences`: a global minimum number of occurrences for
-      each of the `values`. Default: 1 for `all` and `all_and_only`
-      rules, 0 for `only`.
-    * `max_occurrences`: a global maximum number of occurrences for
-      each of the `values`. Default: `inf` (unbounded).
-    * `occurrences_per_value`: a list of dictionaries overriding the
-      global boundaries. Each dictionary may have the following keys:
-
-        * `values` (required): a value (or a list) to which the
-          occurrence bounds below must apply to.
-        * `min_occurrences`: a minimum number of occurrences for these
-          values. Default: global `min_occurrences` parameter.
-        * `max_occurrences`: a maximum number of occurrences for these
-          values. Default: global `max_occurrences` parameter.
-
-      Global parameters apply to all values not present in any of the
-      dictionaries in `occurrences_per_value` (but yet present on the
-      main `values` list).
-
-    * `verbose`: if `True`, the report will include the percentage of
-      occurrences for each value. Default: `True`.
-
-    The `all` rule checks if all the `values` declared are present in
-    the scope (possibly tolerating other values not declared in
-    `values`).
-    Use it when you want to be sure that your data contains at least
-    all the values you declare, also setting `min_occurrences` and
-    `max_occurrences` when necessary.
-    You may also check for "zero" occurrences of a set of values by
-    setting `max_occurrences` to 0.
-
-    The `only` rule ensures that the `values` are the only possible
-    values in the scope (possibly not containing them at all).
-    Use it when you want to enumerate the admitted values for the
-    scope, as in an enumeration.
-
-    The `all_and_only` rule checks both if all the `values` declared
-    are present in the scope and if only they are present (not
-    tolerating values not declared).
-    Use it when you know all the possible values for the scope and you
-    are sure that they will be always present.
-
-    The `min_occurrences` and `max_occurrences` parameters are applied
-    applied to all the `values` declared, and only these. It means you
-    cannot (yet) specify boundaries for values you did't declare.
-
-    You may also notice that, by tweaking the expected number of
-    occurrences, you may end up having the very same behaviour
-    regardless the `rule` you choose.
-    In this case, you should go for the rule that semantically matches
-    best your intents, so that your final validation document looks
-    more readable and easy to understand.
-
-    Examples
-    --------
-    * You have a table of users containg a column `handedness`
-      only admitting the values:
-      `right-handed`, `left-handed` and `ambidextrous`.
-      You know that some of these values may not appear in the data,
-      but you don't want other values to be present.
-
-    .. code-block:: json
-
-        {
-            "scope": "handedness",
-            "statements": [
-                {
-                    "type": "contain",
-                    "rule": "only",
-                    "values": ["right-handed", "left-handed", "ambidextrous"],
-                    "parser": {"dtype": "string"}
-                }
-            ]
-        }
-
-    * You have a table of servers containg a column `role` which may
-      contain the values `master` and `slave`.
-      You want to be sure that there is always one and only one master
-      server in the data.
-
-    .. code-block:: json
-
-        {
-            "scope": "role",
-            "statements": [
-                {
-                    "type": "contain",
-                    "rule": "all",
-                    "values": ["master"],
-                    "parser": {"dtype": "string"},
-                    "min_occurrences": 1,
-                    "max_occurrences": 1
-                }
-            ]
-        }
-
-    You may also extend the previous example by making some adjustments
-    to ensure that there is no other value than `master` and `role`
-    in the data. Make notice that although the `rule` below is changed
-    to `only`, the statement above is still contemplated by the
-    `occurrences_per_value` parameter in the following validation item:
-
-    .. code-block:: json
-
-        {
-            "scope": "role",
-            "statements": [
-                {
-                    "type": "contain",
-                    "rule": "only",
-                    "values": ["master", "slave"],
-                    "parser": {"dtype": "string"},
-                    "occurrences_per_value": [
-                        {
-                            "values": ["master"],
-                            "min_occurrences": 1,
-                            "max_occurrences": 1
-                        }
-                    ]
-                }
-            ]
-        }
-
-    * You have a table of transactions containing details about
-      transactions in all the branches of a company. You expect that
-      there should always be at least one transaction per branch.
-
-    .. code-block:: json
-
-        {
-            "scope": ["branch_name"],
-            "statements": [
-                {
-                    "type": "contain",
-                    "rule": "all_and_only",
-                    "values": [
-                        "Albany", "Utica", "Scranton", "Akron",
-                        "Nashua", "Buffalo", "Rochester"
-                    ],
-                    "parser": {"dtype": "string"}
-                }
-            ]
-        }
-
-    * You have a table for the logs of user accesses to a website which
-      contains an `IP` column. You want to be sure that blacklisted
-      IPs are not present in the data. The following validation item in
-      YAML format checks for the absense of blacklisted IPs:
-
-    .. code-block:: yaml
-
-        scope: IP
-        statements:
-        - type: contain
-          rule: all
-          max_occurrences: 0
-          values: # blacklisted IPs
-          - 3.48.48.135
-          - 3.48.48.136
-          parser: {dtype: string}
-
-    """
-    name = 'contain'
-    expected_parameters = [
-        'rule',
-        'values',
-        'parser',
-        'min_occurrences',
-        'max_occurrences',
-        'occurrences_per_value',
-        'verbose'
-    ]
-    table_only = False
-
-    def __init__(self, *args, **kwargs) -> None:
-        super().__init__(*args, **kwargs)
-
-        self.rule = self.options['rule']
-        self.treater = get_treater_instance(self.options['parser'])
-        self.values = self.treater(self.options['values'])
-
-        self.min_occurrences = self.options.get('min_occurrences', None)
-        self.max_occurrences = self.options.get('max_occurrences', None)
-        self.occurrences_per_value = self.options.get(
-            'occurrences_per_value', []
-        )
-        self.verbose = self.options.get('verbose', True)
-
-        self._set_default_minmax_occurrences()
-        self._assert_parameters()
-
-    def _set_default_minmax_occurrences(self) -> None:
-        min_occurrences_rule_default = {
-            'all': 1,
-            'only': 0,
-            'all_and_only': 1
-        }
-        max_occurrences_rule_default = {
-            'all': inf,
-            'only': inf,
-            'all_and_only': inf
-        }
-
-        if self.min_occurrences is None:
-            self.min_occurrences = min_occurrences_rule_default[self.rule]
-        if self.max_occurrences is None:
-            self.max_occurrences = max_occurrences_rule_default[self.rule]
-
-    def _assert_parameters(self) -> None:
-        assert self.rule in ('all', 'only', 'all_and_only')
-        assert self.min_occurrences >= 0
-        assert self.max_occurrences >= 0
-
-    # docstr-coverage:inherited
-    def report(self, df: DataFrame) -> dict:
-        # Concat all columns
-        count_isin = (
-            concat(df[col] for col in df.columns).value_counts()
-        )
-        self.value_count = count_isin.to_dict()
-
-        # Generate report
-        values = self.treater.serialize(self.value_count.keys())['values']
-        freqs = [int(freq) for freq in self.value_count.values()]
-
-        if self.verbose:
-            # Include percentage in reports
-            total = int(count_isin.sum())
-            rel_freqs = (freq*100/total for freq in freqs)
-            values_report = [
-                {'value': value, 'count': freq, 'perc': pfreq}
-                for value, freq, pfreq in zip(values, freqs, rel_freqs)
-            ]
-        else:
-            values_report = [
-                {'value': value, 'count': freq}
-                for value, freq in zip(values, freqs)
-            ]
-        return {
-            'values': values_report
-        }
-
-    # docstr-coverage:inherited
-    def result(self, report: dict) -> bool:
-        self._set_min_max_boundaries()
-        self._set_values_scope()
-
-        if not self._check_interval(self.value_count):
-            return False
-        if not self._check_rule(self.value_count):
-            return False
-        return True
-
-    def _set_min_max_boundaries(self) -> None:
-        # Global boundaries
-        min_max_boundaries = {}
-        for value in self.values:
-            min_max_boundaries.update({
-                value: {
-                    'min_occurrences': self.min_occurrences,
-                    'max_occurrences': self.max_occurrences
-                }
-            })
-
-        # Dedicated boundaries
-        if self.occurrences_per_value:
-            for occurrence in self.occurrences_per_value:
-                values = occurrence['values']
-                values = [values] if not isinstance(values, list) else values
-                values = self.treater(values)
-
-                for value in values:
-                    min_max_boundaries[value][
-                        'min_occurrences'
-                    ] = occurrence.get(
-                        'min_occurrences', self.min_occurrences
-                    )
-
-                    min_max_boundaries[value][
-                        'max_occurrences'
-                    ] = occurrence.get(
-                        'max_occurrences', self.max_occurrences
-                    )
-
-        self.min_max_boundaries = min_max_boundaries
-
-    def _set_values_scope(self):
-        """
-        Set the scope of values to be analyzed according to the given
-        `self.rule`. Excludes the cases of values where its
-        corresponding `max_occurrences` is zero, since these cases
-        won't matter for the `rule` analysis, as they must be absent
-        in the column.
-        """
-        values_col = [
-            value for value in self.min_max_boundaries
-            if self.min_max_boundaries[value]['max_occurrences'] != 0
-        ]
-        self.values_scope_filter = values_col
-
-    def _check_interval(self, value_count: dict) -> bool:
-        """
-        Check if each value is inside an interval of min and max
-        number of occurrencies. These values are set globally in
-        `self.min_occurrencies` and `self.max_occurrencies`, but the
-        user can specify dedicated intervals for a given value in
-        `self.occurrences_per_value`
-        """
-        for value in self.values:
-            min_value = self.min_max_boundaries[value][
-                'min_occurrences'
-            ]
-            max_value = self.min_max_boundaries[value][
-                'max_occurrences'
-            ]
-            if value in value_count:
-                if not (
-                    min_value <= value_count[value] <= max_value
-                ):
-                    return False
-            else:
-                if self.rule != 'only' and max_value != 0 and min_value != 0:
-                    return False
-        return True
-
-    def _check_rule(self, value_count: dict) -> bool:
-        """
-        Checks if given columns attend the given requirements
-        of presence or absence of values, according to a criteria
-        specified in `self.rule`
-
-        Parameters
-        ----------
-        value_count: dict
-            Got from `report` method, it contains the count of
-            occurrences for each column for each value
-
-        Notes
-        -----
-        `self.rule` parameter defines the criteria to use for checking
-        the presence or absence of values in a column. Its values
-        should be:
-
-        * all: all the values in `self.values` are present in the
-        column, but there can be other values also
-        * only: only the values in `self.values` (but not necessarilly
-        all of them) are present in the given column
-        * all_and_only: the column must contain exactly the values in
-          `self.values` - neither more than less. As the name says, it
-          is an `and` boolean operation between `all` and `only` modes
-        """
-        if self.rule == 'all':
-            return self._check_all(value_count)
-        elif self.rule == 'only':
-            return self._check_only(value_count)
-        else:
-            return (self._check_all(value_count) and
-                    self._check_only(value_count))
-
-    def _check_all(self, value_count: dict) -> bool:
-        """
-        Checks if values in df contains all the expected values
-        """
-        values_in_col = set(value_count.keys())
-        values = set(self.values_scope_filter)
-        if values - values_in_col:
-            return False
-        return True
-
-    def _check_only(self, value_count: dict) -> bool:
-        """
-        Checks if all values in df are inside the expected values
-        """
-        values_in_col = set(value_count.keys())
-        values = set(self.values_scope_filter)
-        if values_in_col - values:
-            return False
-        return True
-
-    # docstr-coverage:inherited
-    @staticmethod
-    def profile(df: DataFrame) -> DeirokayStatement:
-        if any(dtype != df.dtypes for dtype in df.dtypes):
-            raise NotImplementedError(
-                "Refusing to mix up different types of columns"
-            )
-
-        series = concat(df[col] for col in df.columns)
-
-        unique_series = series.drop_duplicates().dropna()
-        if len(unique_series) > 20:
-            raise NotImplementedError("Won't generate too long statements!")
-
-        value_frequency = series.value_counts()
-        min_occurrences = int(value_frequency.min())
-
-        statement_template = {
-            'type': 'contain',
-            'rule': 'all'
-        }  # type: DeirokayStatement
-        # Get most common type to infer treater
-        try:
-            statement_template.update(
-                get_dtype_treater(unique_series.map(type).mode()[0])
-                .serialize(unique_series)
-            )
-        except TypeError:
-            raise NotImplementedError("Can't handle mixed types")
-        statement_template['values'].sort(key=lambda x: (x is None, x))
-
-        if min_occurrences != 1:
-            statement_template['min_occurrences'] = min_occurrences
-
-        return statement_template
diff --git a/deirokay/statements/loader.py b/deirokay/statements/loader.py
new file mode 100644
index 0000000..b96c96d
--- /dev/null
+++ b/deirokay/statements/loader.py
@@ -0,0 +1,88 @@
+import functools
+from types import ModuleType
+from typing import Type
+
+from deirokay._typing import DeirokayStatement
+from deirokay.enums import Backend
+from deirokay.fs import fs_factory
+from deirokay.statements import BaseStatement
+
+from . import STATEMENTS_MAP
+
+
+@functools.lru_cache(maxsize=None)
+def _cached_import_file_as_module(file_path: str) -> ModuleType:
+    """Import a file as a module, caching the result.
+    This prevents the need to read the same file multiple times.
+    """
+    fs = fs_factory(file_path)
+    module = fs.import_as_python_module()
+    return module
+
+
+def _load_custom_statement(location: str) -> Type[BaseStatement]:
+    """Load a custom statement from a .py file"""
+    if '::' not in location:
+        raise ValueError('You should pass your class location using the'
+                         ' following pattern:\n'
+                         '<.py file location>::<class name>')
+
+    file_path, class_name = location.split('::')
+
+    module = _cached_import_file_as_module(file_path)
+    cls = getattr(module, class_name)
+
+    if not issubclass(cls, BaseStatement):
+        raise ImportError('Your custom statement should be a subclass of'
+                          ' BaseStatement')
+
+    return cls
+
+
+def statement_factory(statement: DeirokayStatement,
+                      backend: Backend) -> BaseStatement:
+    """Receive statement dict and create the proper statement object.
+    The `name` attribute of the class is used to bind the statement
+    type to its class.
+
+    Parameters
+    ----------
+    statement : dict
+        Dict from Validation Document representing statement and its
+        parameters.
+
+    Returns
+    -------
+    BaseStatement
+        Instance of Statement class.
+
+    Raises
+    ------
+    KeyError
+        Custom statement should present a `location` default parameter.
+    NotImplementedError
+        Declared statement type does not exist.
+    """
+    stmt_type = statement.get('type')
+
+    if stmt_type == 'custom':
+        location = statement.get('location')
+        if not location:
+            raise KeyError('A custom statement must define a `location`'
+                           ' parameter.')
+        cls = _load_custom_statement(location)
+
+    elif stmt_type in STATEMENTS_MAP:
+        cls = STATEMENTS_MAP[stmt_type]
+
+    else:
+        raise NotImplementedError(
+            f'Statement type "{stmt_type}" not implemented.\n'
+            f'The available types are {list(STATEMENTS_MAP)}'
+            ' or `custom` for your own statements.'
+        )
+
+    execution_class = cls.attach_backend(backend)
+    statement_instance = execution_class(statement)  # type: BaseStatement
+
+    return statement_instance
diff --git a/deirokay/statements/multibackend.py b/deirokay/statements/multibackend.py
new file mode 100644
index 0000000..56b272b
--- /dev/null
+++ b/deirokay/statements/multibackend.py
@@ -0,0 +1,46 @@
+"""Define statement methods that can be aliased in function of the
+backend in use."""
+
+from deirokay._typing import GeneralDecorator
+from deirokay.backend import register_backend_method
+from deirokay.enums import Backend
+
+
+def report(backend: Backend, *, force: bool = False) -> GeneralDecorator:
+    """Define a decorator that turns any method into an alias for the
+    `report` method when executed using a given `backend`.
+
+    Parameters
+    ----------
+    backend : Backend
+        The backend for the decorated method.
+
+    Returns
+    -------
+    GeneralDecorator
+        A decorator for the alias method of `report`.
+    force : bool, optional
+        Force overwrite target method when it already exists.
+        Defaults to False.
+    """
+    return register_backend_method('report', backend, force=force)
+
+
+def profile(backend: Backend, *, force: bool = False) -> GeneralDecorator:
+    """Define a decorator that turns any method into an alias for the
+    `profile` method when executed using a given `backend`.
+
+    Parameters
+    ----------
+    backend : Backend
+        The backend for the decorated method.
+
+    Returns
+    -------
+    GeneralDecorator
+        A decorator for the alias method of `profile`.
+    force : bool, optional
+        Force overwrite target method when it already exists.
+        Defaults to False.
+    """
+    return register_backend_method('profile', backend, force=force)
diff --git a/deirokay/utils.py b/deirokay/utils.py
deleted file mode 100644
index a59f883..0000000
--- a/deirokay/utils.py
+++ /dev/null
@@ -1,42 +0,0 @@
-from difflib import get_close_matches
-from typing import Iterable
-
-from jinja2.nativetypes import Environment
-
-
-def _check_columns_in_df_columns(columns: Iterable, df_columns: Iterable):
-    miss = {}
-    for col in columns:
-        if col not in df_columns:
-            match = get_close_matches(col, df_columns, n=1, cutoff=0)[0]
-            miss[col] = match
-    if miss:
-        raise KeyError(f'Columns {list(miss.keys())} not found in your data.'
-                       f' Did you mean {list(miss.values())}?')
-
-
-def _render_list(env: Environment, list_: list, template: dict):
-    """Render Jinja templates in list recursively."""
-    for index, value in enumerate(list_):
-        if isinstance(value, str) and '{{' in value:
-            rendered = env.from_string(value).render(**template)
-            list_[index] = rendered
-        elif isinstance(value, dict):
-            _render_dict(env, value, template)
-        elif isinstance(value, list):
-            _render_list(env, value, template)
-
-
-def _render_dict(env: Environment, dict_: dict, template: dict):
-    """Render Jinja templates in dict recursively.
-
-    It will render only strings starting with `{{` to prevent
-    unattended renderings."""
-    for key, value in dict_.items():
-        if isinstance(value, str) and '{{' in value:
-            rendered = env.from_string(value).render(**template)
-            dict_[key] = rendered
-        elif isinstance(value, dict):
-            _render_dict(env, value, template)
-        elif isinstance(value, list):
-            _render_list(env, value, template)
diff --git a/deirokay/validator.py b/deirokay/validator.py
index 9022e65..cbbc439 100644
--- a/deirokay/validator.py
+++ b/deirokay/validator.py
@@ -2,110 +2,39 @@
 Set of functions related to Deirokay validation.
 """
 
-import functools
 import json
 import warnings
 from copy import deepcopy
 from datetime import datetime
 from os.path import splitext
-from types import ModuleType
 from typing import Optional, Union
 
-import pandas
 from jinja2 import BaseLoader
 from jinja2 import StrictUndefined as strict
 from jinja2.nativetypes import NativeEnvironment
 
+from deirokay._utils import check_columns_in_df_columns, render_dict
+from deirokay.backend import detect_backend
 from deirokay.enums import SeverityLevel
 from deirokay.exceptions import ValidationError
 from deirokay.fs import FileSystem, LocalFileSystem, fs_factory
 from deirokay.history_template import get_series
-from deirokay.statements import STATEMENTS_MAP, BaseStatement
-from deirokay.utils import _check_columns_in_df_columns, _render_dict
-
-
-@functools.lru_cache(maxsize=32)
-def _cached_import_file_as_module(file_path: str) -> ModuleType:
-    """Import a file as a module, caching the result.
-    This prevents the need to import the same file multiple times.
-    """
-    fs = fs_factory(file_path)
-    module = fs.import_as_python_module()
-    return module
-
-
-def _load_custom_statement(location: str):
-    """Load a custom statement from a .py file"""
-    if '::' not in location:
-        raise ValueError('You should pass your class location using the'
-                         ' following pattern:\n'
-                         '<.py file location>::<class name>')
-
-    file_path, class_name = location.split('::')
-
-    module = _cached_import_file_as_module(file_path)
-    cls = getattr(module, class_name)
-
-    if not issubclass(cls, BaseStatement):
-        raise ImportError('Your custom statement should be a subclass of '
-                          'BaseStatement')
-
-    return cls
-
-
-def _process_stmt(statement: dict) -> BaseStatement:
-    """Receive statement dict and call the proper statement class.
-    The `name` attribute of the class is used to bind the statement
-    type to its class.
-
-    Parameters
-    ----------
-    statement : dict
-        Dict from Validation Document representing statement and its
-        parameters.
-
-    Returns
-    -------
-    BaseStatement
-        Instance of Statement class.
-
-    Raises
-    ------
-    KeyError
-        Custom statement should present a `location` default parameter.
-    NotImplementedError
-        Declared statement type does not exist.
-    """
-    stmt_type = statement.get('type')
-
-    if stmt_type == 'custom':
-        location = statement.get('location')
-        if not location:
-            raise KeyError('A custom statement must define a `location`'
-                           ' parameter.')
-        cls = _load_custom_statement(location)
-    elif stmt_type in STATEMENTS_MAP:
-        cls = STATEMENTS_MAP[stmt_type]
-    else:
-        raise NotImplementedError(
-            f'Statement type "{stmt_type}" not implemented.\n'
-            f'The available types are {list(STATEMENTS_MAP)}'
-            ' or `custom` for your own statements.'
-        )
-
-    statement_instance: BaseStatement = cls(statement)
-
-    return statement_instance
-
-
-def validate(df: pandas.DataFrame, *,
-             against: Union[str, dict],
-             save_to: Optional[str] = None,
-             save_format: str = None,
-             current_date: Optional[datetime] = None,
-             raise_exception: bool = True,
-             exception_level: SeverityLevel = SeverityLevel.CRITICAL,
-             template: Optional[dict] = None) -> dict:
+from deirokay.statements.loader import statement_factory
+
+from ._typing import (DeirokayDataSource, DeirokayValidationDocument,
+                      DeirokayValidationResultDocument)
+
+
+def validate(
+    df: DeirokayDataSource, *,
+    against: Union[str, DeirokayValidationDocument],
+    save_to: Optional[str] = None,
+    save_format: str = None,
+    current_date: Optional[datetime] = None,
+    raise_exception: bool = True,
+    exception_level: SeverityLevel = SeverityLevel.CRITICAL,
+    template: Optional[dict] = None
+) -> DeirokayValidationResultDocument:
     """Validate a Deirokay DataFrame against a well-defined Validation
     Document.
 
@@ -157,8 +86,8 @@ def validate(df: pandas.DataFrame, *,
 
     Returns
     -------
-    dict
-        Validation Result Document.
+    DeirokayValidationResultDocument
+        Validation Result Document dict.
 
     Raises
     ------
@@ -168,6 +97,7 @@ def validate(df: pandas.DataFrame, *,
         Validation failed for at least one statement whose severity is
         greater or equal to `exception_level`.
     """
+    backend = detect_backend(df)
 
     if save_to:
         save_to_fs = fs_factory(save_to)
@@ -190,19 +120,19 @@ def validate(df: pandas.DataFrame, *,
         series=lambda x, y: get_series(x, y, read_from=save_to_fs),
         **(template or {})
     )
-    _render_dict(NativeEnvironment(loader=BaseLoader(), undefined=strict),
-                 dict_=validation_document,
-                 template=template)
+    render_dict(NativeEnvironment(loader=BaseLoader(), undefined=strict),
+                dict_=validation_document,
+                template=template)
 
     for item in validation_document['items']:
-        scope = item.get('scope')
+        scope = item['scope']
         scope = [scope] if not isinstance(scope, list) else scope
-        _check_columns_in_df_columns(scope, df.columns)
+        check_columns_in_df_columns(scope, df.columns)
 
         df_scope = df[scope]
 
         for stmt in item.get('statements'):
-            report = _process_stmt(stmt)(df_scope)
+            report = statement_factory(stmt, backend)(df_scope)
             if report['result'] is True:
                 report['result'] = 'pass'
             else:
@@ -241,9 +171,9 @@ def raise_validation(validation_result_document: dict,
     highest_level = None
     for item in validation_result_document['items']:
         scope = item['scope']
-        for stmt in item.get('statements'):
+        for stmt in item['statements']:
             severity = stmt.get('severity', SeverityLevel.CRITICAL)
-            result = stmt.get('report').get('result')
+            result = stmt['report']['result']
 
             if result == 'fail':
                 if severity >= exception_level:
diff --git a/docs/source/concepts/definitions.rst b/docs/source/concepts/definitions.rst
index 024512e..4ab4bfc 100644
--- a/docs/source/concepts/definitions.rst
+++ b/docs/source/concepts/definitions.rst
@@ -163,9 +163,9 @@ Once you have correctly use `deirokay.data_reader` to parse your data into a Dat
 .. code-block:: python
 
     from deirokay import data_reader, profile
+    from deirokay.enums import Backend
 
-
-    df = data_reader('file.csv', options='options.json')
+    df = data_reader('file.csv', options='options.json', backend=Backend.PANDAS)
 
     profile(df, 'CUSTOMERS', save_to='validation_doc.json')
 
diff --git a/docs/source/concepts/statements.rst b/docs/source/concepts/statements.rst
index 992c848..6d11876 100644
--- a/docs/source/concepts/statements.rst
+++ b/docs/source/concepts/statements.rst
@@ -72,14 +72,14 @@ several additional parameters, and they can be optional or not.
 Special Parameters for Statements
 =================================
 
-There are special parameter names, which are actually directives to 
+There are special parameter names, which actually are directives to 
 pass attributes regarding the statement itself, and not the statement 
 type. They are:
 
 - **type**: It defines the statement type itself, so it cannot be used 
   as a parameter name. Should be set to *custom* in case of custom 
-  statements, and the statement class location should be declared in 
-  **location**.
+  statements. In this case, the statement class location should be
+  declared using the **location** attribute.
 
 - **severity**: Set a severity level for a statement in case of failure.
   It must be an integer value, conventionally from 1 (minimal) to 5 
@@ -98,7 +98,9 @@ For a statement to be processed, its type, as declared in the
 validation document, must match a statement class name (the *name* 
 attribute of a statement class).
 
-The native statement classes are all declared in :ref:`deirokay.statements`. See an example:
+The native statement classes are all declared in
+:ref:`deirokay.statements`. See a simplified version of the `not_null`
+statement:
 
 .. code-block:: python
 
@@ -150,36 +152,100 @@ The native statement classes are all declared in :ref:`deirokay.statements`. See
             }
             return statement
 
-When processing statements, Deirokay will list all class names in 
-this module and look for a class whose name matches the *type* key from 
-the validation document. It means it is enough to create a class in 
-this module to make it available to use.
-
-The *expected_parameters* is a mandatory argument to identify all valid 
-parameters for the current class. Any parameter that is neither special 
-nor listed as expected will raise an exception.
-
-Ideally, the statement class could validate the parameters it accepts 
-in the *__init__* method.
-
-The *report* method is intended to report statistics that may be useful 
+When processing statements, Deirokay will decide which statement class
+to load based on the `type` parameter declared in the Validation
+Document. This parameter should coincide with the `name` attribute of
+one of the native statement classes. Custom statement classes should
+still have a `name` class attribute, but, as previously stated, they
+should be signaled as an external dependency to Deirokay by declaring
+`type: 'custom'` and a valid `location` parameter in the Validation
+Document.
+
+The `expected_parameters` is a mandatory attribute for Deirokay to
+identify all valid parameters for the current class. Any parameter that
+is neither special nor listed as expected will raise an exception.
+Ideally, the statement class should validate each of its custom
+parameters during initialization (`__init__` method).
+
+The `report` method is intended to report statistics that may be useful 
 for the current statement. Thinking of the validation process also as a 
 form of logging (when the validation result is saved), the metrics 
 reported by the statement could be useful in a numerous use cases. 
-Ideally, the *report* method should also summarize all calculations 
+Ideally, the `report` method should also summarize all calculations 
 that will be logically evaluated by the next method.
 
-The *result* method has only one purpose: return either True (for a 
-successful test) or False (for a failed test). A failure can be a 
+The `result` method has only one purpose: return either `True` (for a 
+successful test) or `False` (for a failed test). A failure can be a 
 consequence of several reasons, since a statement is able to evaluate a 
 series of parameters passed by the user and a set of metrics reported 
-by the *report* method.
+by the `report` method.
 
-The *profile* is a static method used to generate a default statement 
+The `profile` is a static method used to generate a default statement 
 object for the current class. It is not called during the validation 
 process, but when profiling the data. When the user calls the 
-*deirokay.profile* function, all native statement classes having a 
-*profile* method are iterated to generate a default statement. By 
+`deirokay.profile` function, all native statement classes having a 
+`profile` method are iterated over to generate a default statement. By 
 default, statement objects are generated for the entire template 
 DataFrame (the entire set of columns), and then for each of its columns 
 individually.
+
+Dealing with multi-backend resources
+====================================
+
+Since Deirokay 1.0, statement classes also support multiple engines for
+statement evaluation. Every class should declare a `supported_backends`
+attribute containing the objects from the `Backend` enum corresponding
+to the backends it supports.
+
+Currently, only two methods support "overloading" for distinct backends:
+`report` and `profile`.
+Two homonymous decorators used to indicate the implementation for each
+supported backend are
+`@deirokay.statements.multibackend.report` and
+`@deirokay.statements.multibackend.profile`. 
+
+See an example:
+
+.. code-block:: python
+
+    from deirokay.statements import BaseStatement
+    from deirokay.statements.multibackend import report, profile
+    from deirokay.enums import Backend
+
+
+    class DummyMultibackendStatement(BaseStatement):
+        name = 'dummy'
+        expected_parameters = []
+        supported_backends = [Backend.PANDAS, Backend.DASK]
+
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+
+        @report(Backend.PANDAS)
+        def _report_method_for_pandas_backend(self, df):
+            ...
+
+        @report(Backend.DASK)
+        def _report_method_for_dask_backend(self, df):
+            ...
+
+        def result(self, report):
+            ...
+
+        @profile(Backend.PANDAS)
+        @staticmethod
+        def _profile_for_pandas(df):
+            ...
+
+        @profile(Backend.DASK)
+        @staticmethod
+        def _profile_for_dask(df):
+            ...
+
+
+Make notice both `report`-decorated should generate the exact outputs,
+as the `result` method cannot be overloaded for distinct backends (this
+is done on purpose). Also remark that the `@profile` decorator should
+be on top of `@staticmethod`, as the `staticmethod` descriptor will
+also be copied to the target `profile` method during the statement
+evaluation phase.
diff --git a/docs/source/getting_started.rst b/docs/source/getting_started.rst
index 0c2b6d3..2978596 100644
--- a/docs/source/getting_started.rst
+++ b/docs/source/getting_started.rst
@@ -110,7 +110,8 @@ pandas dataframe that doesn't have the initial problems:
 .. code-block:: python
 
     >>> from deirokay import data_reader
-    >>> data_reader('example.csv', options='options.json')
+    >>> from deirokay.backend import Backend
+    >>> data_reader('example.csv', options='options.json', backend=Backend.PANDAS)
          name   age  is_married
     0    John    55        True
     1  Mariah    44        True
@@ -215,8 +216,11 @@ Finally, to test your dataset against the validation document:
 .. code-block:: python
 
     from deirokay import data_reader
+    from deirokay.backend import Backend
 
-    df = data_reader('example.csv', options='options.json')
+    df = data_reader('example.csv',
+                     options='options.json',
+                     backend=Backend.PANDAS)
     validation_result_document = validate(df,
                                           against='assertions.json',
                                           raise_exception=False)
diff --git a/docs/source/index.rst b/docs/source/index.rst
index eae042a..c2e74e7 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -25,12 +25,12 @@ Deirokay comes with several features to help you in your daily
 data quality problems. The main tools that facilitate your job
 are *Data Reader*, *Data Validation* and *Data Profiling*.
 
-The :ref:`Data Reader` is responsible for parsing and transforming your source data into a trustable DataFrame before it gets validated. You may get a DataFrame from a file, a database or an existing Pandas DataFrame that has not been treated yet, all you need to do is to provide a Deirokay Options Document specifying the parsing options.
-Deirokay Data Reader can also be imported and used in your project to improve your analysis and data manipulation when using Pandas.
+The :ref:`Data Reader` is responsible for parsing and transforming your source data into a trustable DataFrame before it gets validated. You may get a DataFrame from a file, a database or an existing DataFrame that has not been treated yet; all you need to do is to provide a Deirokay Options Document specifying the parsing options.
+Deirokay Data Reader can also be imported and used in your project to improve your analysis and data manipulation when using Pandas or any of the supported backends.
 
 
 Data validation is performed by the :ref:`Data Validator` and works with the help of :ref:`Deirokay Validation Documents<Validation Document>`. The latter must be created for each data file and they gather all the validation (using native or custom statements) to be executed against the data. 
-Every validation results in a *Validation Result Document*, which you may choose to log for reference or later use.
+Every validation results in a *Validation Result Document*, which you may choose to log for reference or later use. As for the Data Reader, the validation may use different backends, such as Pandas and Dask. 
 
 :ref:`Data Profiling<Data Profiling: Auto-generated Validation Document>` is provided by the :ref:`deirokay.profile<deirokay.profiler.profile>` method and is here to help you with automatically generating Validation Documents based on template data. 
 The user is encouraged to modify and supplement the generated document to better meet their requirements.
diff --git a/docs/source/installation.rst b/docs/source/installation.rst
index 7554a3d..ae938f3 100644
--- a/docs/source/installation.rst
+++ b/docs/source/installation.rst
@@ -3,22 +3,34 @@ Installation
 
 To start using Deirokay, install its package by following the instructions below.
 
-Install Deirokay using the pip command:
+By default, only core dependencies are installed when supplying the pip
+command:
 
 .. code-block:: bash
 
     pip install Deirokay
 
-To include optional dependencies for AWS S3, install:
+Depending on your use cases, you need to include extra dependencies.
+Use any of the commands below:
 
 .. code-block:: bash
 
-    pip install Deirokay[s3]
+    pip install Deirokay[s3]  # Optional dependencies for use with AWS S3
+    pip install Deirokay[pandas]  # For Pandas backend
+    pip install Deirokay[dask]  # For Dask or Dask Distributed backends
 
+You may also install any combination of the extras above by separating
+them with commas:
 
-If you wish to contribute for Deirokay development, maybe
-you will want to install a more complete set of packages for
-testing to help you in your development.
+.. code-block:: bash
+
+    pip install Deirokay[dask,s3]  # Dask + S3 deps
+
+
+If you wish to contribute for Deirokay development, you need to install
+a more restrict set of packages, in order to guarantee that you are
+always creating and testing code to work with least recent supported
+versions:
 
 .. code-block:: bash
 
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 5194468..2c6eda7 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,17 +1,21 @@
 # Force minimum requirements
 Jinja2==2.10.2
+markupsafe==1.1.1 ; python_version <= "3.7"
 jq==1.2.1
 pandas==1.3 ; python_version < "3.10"
 pandas==1.4 ; python_version >= "3.10"
-PyYAML==5.1
+PyYAML==5.3.1
+lazy_import==v0.2.2
+dask[dataframe]==2022.1.1
 # Libraries for testing
 apache-airflow==2.0.0
 boto3
 moto
-psycopg
 pyarrow
 pytest
 pytest-cov
-pytest-postgresql
+pytest-postgresql  # requires postgresql
+psycopg[binary]  # for `pytest-postgresql`
+psycopg2[binary]  # for `pandas.read_sql` from connection string
 s3fs
 SQLAlchemy
diff --git a/requirements-s3.txt b/requirements-s3.txt
deleted file mode 100644
index 3f5c76e..0000000
--- a/requirements-s3.txt
+++ /dev/null
@@ -1,2 +0,0 @@
-boto3
-s3fs
diff --git a/requirements.txt b/requirements.txt
index f1fd823..23461a5 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,4 @@
 Jinja2<3,>=2.10.2
 jq<2,>=1.2.1
-pandas<2,>=1.3
-PyYAML<7,>=5.1
+PyYAML<7,>=5.3.1
+lazy_import>=v0.2.2
diff --git a/setup.py b/setup.py
index ea9fd45..c075a10 100644
--- a/setup.py
+++ b/setup.py
@@ -15,9 +15,6 @@
 with open('requirements-dev.txt') as f:
     requirements_dev = [line.strip() for line in f.readlines()]
 
-with open('requirements-s3.txt') as f:
-    requirements_s3 = [line.strip() for line in f.readlines()]
-
 setup(
     name="deirokay",
     packages=find_packages(include=['deirokay*']),
@@ -38,6 +35,8 @@
     install_requires=requirements,
     extras_require={
         'dev': requirements_dev,
-        's3': requirements_s3
+        's3': ['boto3', 's3fs'],
+        'pandas': ['pandas<2,>=1.3'],
+        'dask': ['dask[dataframe]>=2022.1.1'],
     }
 )


Test patch:
diff --git a/tests/custom_statement.py b/tests/custom_statement.py
index f036b91..6925577 100644
--- a/tests/custom_statement.py
+++ b/tests/custom_statement.py
@@ -1,21 +1,32 @@
+from deirokay.enums import Backend
 from deirokay.statements import BaseStatement
+from deirokay.statements.multibackend import report
 
 
 class ThereAreValuesGreaterThanX(BaseStatement):
     name = 'there_are_values_greater_than_x'
     expected_parameters = ['x']
+    supported_backends = [Backend.PANDAS, Backend.DASK]
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
         self.x = self.options.get('x')
 
-    def report(self, df):
+    def _report_common(self, df):
         bools = df > self.x
         report = {
             'values_greater_than_x': list(bools[bools.all(axis=1)].index)
         }
         return report
 
+    @report(Backend.PANDAS)
+    def _report_pandas(self, df):
+        return self._report_common(df)
+
+    @report(Backend.DASK)
+    def _report_dask(self, df):
+        return self._report_common(df)
+
     def result(self, report):
         return len(report.get('values_greater_than_x')) > 0
diff --git a/tests/statements/test_column_expression.py b/tests/statements/test_column_expression.py
index f535616..1bfe416 100644
--- a/tests/statements/test_column_expression.py
+++ b/tests/statements/test_column_expression.py
@@ -1,8 +1,10 @@
 import pytest
 
 from deirokay import data_reader, validate
+from deirokay.enums import Backend
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize(
     'scope, expressions, at_least, at_most, result',
     [
@@ -29,10 +31,12 @@
         (['c1', 'c9'], 'c1 + 0.1 == c9', 60, 100, 'pass'),
         (['c1', 'c9'], 'c1 + 0.1 == c9', 61, 100, 'fail')
     ])
-def test_column_expression(scope, expressions, at_least, at_most, result):
+def test_column_expression(scope, expressions, at_least, at_most, result,
+                           backend):
     df = data_reader(
         'tests/statements/test_column_expression.csv',
-        options='tests/statements/test_column_expression_options.yaml'
+        options='tests/statements/test_column_expression_options.yaml',
+        backend=backend
     )
     assertions = {
         'name': 'expressions_test',
@@ -56,6 +60,7 @@ def test_column_expression(scope, expressions, at_least, at_most, result):
     ) == result
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize(
     'scope, expressions, at_least, at_most, result',
     [
@@ -70,10 +75,12 @@ def test_column_expression(scope, expressions, at_least, at_most, result):
         (['c1', 'c8'], 'c1 =~ c8', 100, 100, 'pass'),
         (['c1', 'c9'], 'c1 =~ c9', 60, 100, 'pass')
     ])
-def test_isclose_expression(scope, expressions, at_least, at_most, result):
+def test_isclose_expression(scope, expressions, at_least, at_most, result,
+                            backend):
     df = data_reader(
         'tests/statements/test_column_expression.csv',
-        options='tests/statements/test_column_expression_options.yaml'
+        options='tests/statements/test_column_expression_options.yaml',
+        backend=backend
     )
     assertions = {
         'name': 'expressions_isclose_test',
diff --git a/tests/statements/test_contain.csv b/tests/statements/test_contain.csv
index b469fee..ddcbcda 100644
--- a/tests/statements/test_contain.csv
+++ b/tests/statements/test_contain.csv
@@ -2,10 +2,10 @@ test_rule_1;test_rule_2;test_rule_3;test_maxmin;test_not_contain
 RJ;RJ;RJ;RJ;RJ
 ES;ES;ES;RJ;ES
 SC;SC;SC;RJ;SP
-AC;AC;AC;SP;MG
-SP;;SP;SP;MG
-AB;;AC;ES;SP
-XY;;;ES;SP
-hello_world;;;ES;SP
-hello_deirokay;;;ES;BA
-hello_life;;;ES;BA
+AC;SC;AC;SP;MG
+SP;RJ;;SP;MG
+;ES;AC;ES;SP
+;SC;AC;ES;SP
+;SC;AC;ES;SP
+;SC;AC;ES;BA
+;SC;;ES;BA
diff --git a/tests/statements/test_contain.py b/tests/statements/test_contain.py
index 7d251be..944ca66 100644
--- a/tests/statements/test_contain.py
+++ b/tests/statements/test_contain.py
@@ -1,19 +1,22 @@
 import pytest
 
 from deirokay import data_reader, validate
-from deirokay.statements import Contain
+from deirokay.enums import Backend
+from deirokay.statements.builtin import Contain
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize('rule, scope, result',
                          [('all', 'test_rule_1', 'pass'),
                           ('all', 'test_rule_2', 'fail'),
-                          ('all', 'test_rule_3', 'pass'),
+                          ('all', 'test_rule_3', 'fail'),
                           ('only', 'test_rule_1', 'fail'),
                           ('only', 'test_rule_2', 'pass'),
-                          ('only', 'test_rule_3', 'pass')])
-def test_rules(rule, scope, result):
+                          ('only', 'test_rule_3', 'fail')])
+def test_rules(rule, scope, result, backend):
     df = data_reader('tests/statements/test_contain.csv',
-                     options='tests/statements/test_contain_options.yaml')
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
     assertions = {
         'name': 'all_test_rule',
         'items': [
@@ -36,6 +39,7 @@ def test_rules(rule, scope, result):
     ) == result
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize('occurrences, result',
                          [({'min_occurrences': 3}, 'fail'),
                           ({'min_occurrences': 1}, 'pass'),
@@ -55,7 +59,7 @@ def test_rules(rule, scope, result):
                                   {'values': ['SP'], 'min_occurrences': 2}
                               ]
                           }, 'pass')])
-def test_max_min(occurrences, result):
+def test_max_min(occurrences, result, backend):
     """
     Tests if statement `contain`'s checks about minimum
     and maximum quantities for each column/value are correct.
@@ -63,7 +67,8 @@ def test_max_min(occurrences, result):
     and the `occurrences_per_value` parameter also.
     """
     df = data_reader('tests/statements/test_contain.csv',
-                     options='tests/statements/test_contain_options.yaml')
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
 
     assertions = {
         'name': 'max_min_global_test',
@@ -87,13 +92,15 @@ def test_max_min(occurrences, result):
     ) == result
 
 
-def test_rule_not_contain():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_rule_not_contain(backend):
     """
     Tests the extremal case of not containing some values, obtained
     by combining `rule = 'all'` and `max_occurrences = 0`
     """
     df = data_reader('tests/statements/test_contain.csv',
-                     options='tests/statements/test_contain_options.yaml')
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
     assertions = {
         'name': 'all_not_contain_test_rule',
         'items': [
@@ -117,14 +124,17 @@ def test_rule_not_contain():
     ) == 'pass'
 
 
-def test_profile():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_profile(backend):
     """
     Tests if `profile` method outputs the expected value
     """
     df = data_reader('tests/statements/test_contain.csv',
-                     options='tests/statements/test_contain_options.yaml')
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
 
-    generated_prof = Contain.profile(df[['test_maxmin']])
+    contain = Contain.attach_backend(backend)
+    generated_prof = contain.profile(df[['test_maxmin']])
 
     expected_profile = {
         'type': 'contain',
@@ -147,3 +157,73 @@ def test_profile():
         validate(df, against=assertions, raise_exception=False)
         ['items'][0]['statements'][0]['report']['result']
     ) == 'pass'
+
+
+@pytest.mark.parametrize('backend', list(Backend))
+def test_multicolumn(backend):
+    df = data_reader('tests/statements/test_contain.csv',
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
+    assertions = {
+        'name': 'all_test_rule',
+        'items': [
+            {
+                'scope': ['test_rule_1', 'test_rule_2'],
+                'statements': [
+                    {
+                        'type': 'contain',
+                        'rule': 'all',
+                        'values': [
+                            ['RJ', 'RJ'],
+                            ['ES', 'ES']
+                        ],
+                        'multicolumn': True,
+                        'parsers': 2*[{'dtype': 'string'}],
+                        'occurrences_per_value': [
+                            {
+                                'values': [('RJ', 'RJ'), ('ES', 'ES')],
+                                'min_occurrences': 1,
+                            }
+                        ]
+                    }
+                ]
+            }
+        ]
+    }
+    assert (
+        validate(df, against=assertions, raise_exception=False)
+        ['items'][0]['statements'][0]['report']['result']
+    ) == 'pass'
+
+
+@pytest.mark.parametrize('backend', list(Backend))
+@pytest.mark.parametrize('scope, rule, result, values', [
+    ('test_rule_1', 'all', 'pass', ['AC', 'SP', None]),
+    ('test_rule_1', 'only', 'fail', [None]),
+    ('test_rule_2', 'all', 'fail', [None]),
+    ('test_not_contain', 'all', 'fail', ['RJ', 'SP', 'ES', None]),
+])
+def test_null_values(scope, rule, result, values, backend):
+    df = data_reader('tests/statements/test_contain.csv',
+                     options='tests/statements/test_contain_options.yaml',
+                     backend=backend)
+    assertions = {
+        'name': 'all_test_rule',
+        'items': [
+            {
+                'scope': scope,
+                'statements': [
+                    {
+                        'type': 'contain',
+                        'rule': rule,
+                        'values': values,
+                        'parser': {'dtype': 'string'}
+                    }
+                ]
+            }
+        ]
+    }
+    assert (
+        validate(df, against=assertions, raise_exception=False)
+        ['items'][0]['statements'][0]['report']['result']
+    ) == result
diff --git a/tests/statements/test_not_null.py b/tests/statements/test_not_null.py
index dc1a5d1..fe763a5 100644
--- a/tests/statements/test_not_null.py
+++ b/tests/statements/test_not_null.py
@@ -1,12 +1,17 @@
 from pprint import pprint
 
+import pytest
+
 from deirokay import data_reader, profile, validate
+from deirokay.enums import Backend
 
 
-def test_not_null_statement():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_not_null_statement(backend):
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -36,11 +41,13 @@ def test_not_null_statement():
     validate(df, against=assertions)
 
 
-def test_profile_wont_generate_useless_statement():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_profile_wont_generate_useless_statement(backend):
     """Prevent statement generation when `at_least_%` is 0."""
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     validation_doc = profile(df, 'sample')
diff --git a/tests/statements/test_row_count.py b/tests/statements/test_row_count.py
index f8128af..e9f79d2 100644
--- a/tests/statements/test_row_count.py
+++ b/tests/statements/test_row_count.py
@@ -1,9 +1,11 @@
 import pytest
 
 from deirokay import data_reader, validate
-from deirokay.statements import RowCount
+from deirokay.enums import Backend
+from deirokay.statements.builtin import RowCount
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize(
     'scope, params, result',
     [
@@ -16,9 +18,10 @@
         ('COD_MERC_SERV02', {'distinct': True, 'min': 9, 'max': 9}, 'fail'),
     ]
 )
-def test_row_count(scope, params, result):
+def test_row_count(scope, params, result, backend):
     df = data_reader('tests/transactions_sample.csv',
-                     options='tests/options.yaml')
+                     options='tests/options.yaml',
+                     backend=backend)
     assertions = {
         'name': 'test_row_count',
         'items': [
@@ -38,10 +41,12 @@ def test_row_count(scope, params, result):
     ) == result
 
 
-def test_profile_over_multi_columns():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_profile_over_multi_columns(backend):
     df = data_reader('tests/transactions_sample.csv',
-                     options='tests/options.yaml')
-    doc = RowCount.profile(df)
+                     options='tests/options.yaml',
+                     backend=backend)
+    doc = RowCount.attach_backend(backend).profile(df)
     assert doc == {
         'type': 'row_count',
         'min': 20,
@@ -49,6 +54,7 @@ def test_profile_over_multi_columns():
     }
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize(
     'column, expected',
     [
@@ -72,8 +78,9 @@ def test_profile_over_multi_columns():
         })
     ]
 )
-def test_profile_over_single_column(column, expected):
+def test_profile_over_single_column(column, expected, backend):
     df = data_reader('tests/transactions_sample.csv',
-                     options='tests/options.yaml')
-    doc = RowCount.profile(df[[column]])
+                     options='tests/options.yaml',
+                     backend=backend)
+    doc = RowCount.attach_backend(backend).profile(df[[column]])
     assert doc == expected
diff --git a/tests/statements/test_unique.py b/tests/statements/test_unique.py
index 9d19a36..e7d28e1 100644
--- a/tests/statements/test_unique.py
+++ b/tests/statements/test_unique.py
@@ -1,25 +1,76 @@
-from pprint import pprint
+import pytest
 
-from deirokay import data_reader, profile
+from deirokay import data_reader, validate
+from deirokay.enums import Backend
+from deirokay.statements.builtin import Unique
 
 
-def test_profile_wont_generate_useless_statement():
-    """Prevent statement generation when `at_least_%` is 0."""
+@pytest.mark.parametrize('backend', list(Backend))
+def test_unique(backend):
+    df = data_reader(
+        'tests/transactions_sample.csv',
+        options='tests/options.json',
+        backend=backend
+    )
+    assertions = {
+        'name': 'test_unique',
+        'items': [
+            {
+                'scope': df.columns.tolist(),
+                'statements': [
+                    {
+                        'type': 'unique'
+                    }
+                ]
+            }
+        ]
+    }
+    result = validate(df, against=assertions, raise_exception=False)
+    assert (
+        result['items'][0]['statements'][0]['report']['detail'] == {
+            'unique_rows': 20,
+            'unique_rows_%': 100.0,
+        }
+    )
+
+
+@pytest.mark.parametrize('backend', list(Backend))
+def test_not_unique(backend):
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
+    assertions = {
+        'name': 'test_not_unique',
+        'items': [
+            {
+                'scope': 'PROD_VENDA',
+                'statements': [
+                    {
+                        'type': 'unique'
+                    }
+                ]
+            }
+        ]
+    }
+    result = validate(df, against=assertions, raise_exception=False)
+    assert (
+        result['items'][0]['statements'][0]['report']['detail'] == {
+            'unique_rows': 9,
+            'unique_rows_%': 45.0,
+        }
+    )
+
+
+@pytest.mark.parametrize('backend', list(Backend))
+def test_profile_wont_generate_useless_statement(backend):
+    """Prevent statement generation when `at_least_%` is 0."""
+    df = data_reader(
+        'tests/transactions_sample.csv',
+        options='tests/options.json',
+        backend=backend
+    )[['DT_OPERACAO01']]
 
-    validation_doc = profile(df, 'sample')
-    pprint(validation_doc)
-
-    # Check that the statement was not generated for
-    # "DT_OPERACAO01" column
-    scope_empty = next(filter(
-        lambda x: x['scope'] == 'DT_OPERACAO01',
-        validation_doc['items']
-    ))
-    assert len([
-        stmt for stmt in scope_empty['statements']
-        if stmt['type'] == 'unique'
-    ]) == 0
+    with pytest.raises(NotImplementedError):
+        Unique.attach_backend(backend).profile(df)
diff --git a/tests/test_airflow_operator.py b/tests/test_airflow_operator.py
index cd40958..d62f9d7 100644
--- a/tests/test_airflow_operator.py
+++ b/tests/test_airflow_operator.py
@@ -2,14 +2,17 @@
 from airflow.exceptions import AirflowSkipException
 
 from deirokay.airflow import DeirokayOperator
+from deirokay.enums import Backend
 
 
-def test_deirokay_operator(prepare_history_folder):
+@pytest.mark.parametrize('backend', list(Backend))
+def test_deirokay_operator(prepare_history_folder, backend):
     operator = DeirokayOperator(
         task_id='deirokay_validate',
         data='tests/transactions_sample.csv',
         options='tests/options.json',
         against='tests/assertions_with_history.json',
+        backend=backend,
         template={'forty': 40},
         save_to=prepare_history_folder
     )
@@ -19,7 +22,8 @@ def test_deirokay_operator(prepare_history_folder):
     operator.execute({'ts_nodash': '20001231T101012'})
 
 
-def test_deirokay_operator_with_severity(prepare_history_folder):
+@pytest.mark.parametrize('backend', list(Backend))
+def test_deirokay_operator_with_severity(prepare_history_folder, backend):
 
     assertions = {
         "name": "VENDAS",
@@ -51,6 +55,7 @@ def test_deirokay_operator_with_severity(prepare_history_folder):
         data='tests/transactions_sample.csv',
         options='tests/options.json',
         against=assertions,
+        backend=backend,
         save_to=prepare_history_folder
     )
 
diff --git a/tests/test_data_profiling.py b/tests/test_data_profiling.py
index 92df290..981f8e4 100644
--- a/tests/test_data_profiling.py
+++ b/tests/test_data_profiling.py
@@ -1,13 +1,18 @@
 from pprint import pprint
 
+import pytest
+
 from deirokay import data_reader, profile, validate
+from deirokay.enums import Backend
 
 
-def test_profiling():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_profiling(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     validation_document = profile(df, 'transactions_sample')
diff --git a/tests/test_data_reader.py b/tests/test_data_reader.py
index fa1cbe6..3cac985 100644
--- a/tests/test_data_reader.py
+++ b/tests/test_data_reader.py
@@ -1,36 +1,36 @@
-import psycopg
+import dask.dataframe as dd
+import pandas
+import psycopg2
 import pytest
-from pandas import read_csv
 
 from deirokay import data_reader
-from deirokay.enums import DTypes
+from deirokay.enums import Backend, DTypes
 
 
-def test_data_reader_with_json_options():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_with_json_options(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
     assert len(df) == 20
 
-    print(df)
-    print(df.dtypes)
-
 
-def test_data_reader_with_yaml_options():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_with_yaml_options(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.yaml'
+        options='tests/options.yaml',
+        backend=backend
     )
     assert len(df) == 20
 
-    print(df)
-    print(df.dtypes)
-
 
-def test_data_reader_with_dict_options():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_with_dict_options(backend):
 
     options = {
         'encoding': 'iso-8859-1',
@@ -64,7 +64,8 @@ def test_data_reader_with_dict_options():
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options=options
+        options=options,
+        backend=backend
     )
     assert len(df) == 20
 
@@ -72,7 +73,8 @@ def test_data_reader_with_dict_options():
     print(df.dtypes)
 
 
-def test_data_reader_with_dict_options_only_a_few_columns():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_with_dict_options_only_a_few_columns(backend):
 
     options = {
         'encoding': 'iso-8859-1',
@@ -86,33 +88,34 @@ def test_data_reader_with_dict_options_only_a_few_columns():
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options=options
+        options=options,
+        backend=backend
     )
     assert len(df) == 20
     assert len(df.columns) == 2
     assert all(col in ('WERKS01', 'DT_OPERACAO01') for col in df.columns)
 
 
-def test_data_reader_without_options_exception():
-    with pytest.raises(TypeError):
-        data_reader(
-            'tests/transactions_sample.csv'
-        )
-
-
-def test_data_reader_parquet():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_parquet(backend):
     df = data_reader(
         'tests/sample_parquet.parquet',
-        options='tests/sample_parquet.json'
+        options='tests/sample_parquet.json',
+        backend=backend
     )
+    assert len(df) == 3
 
-    print(df)
-    print(df.dtypes)
 
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_from_dataframe(backend):
+
+    read_csv = {
+        Backend.PANDAS: pandas.read_csv,
+        Backend.DASK: dd.read_csv,
+    }[backend]
 
-def test_data_reader_from_dataframe():
-    df = read_csv('tests/transactions_sample.csv', sep=';',
-                  thousands='.', decimal=',')
+    df = read_csv('tests/transactions_sample.csv',
+                  sep=';', thousands='.', decimal=',')
 
     options = {
         'encoding': 'iso-8859-1',
@@ -139,24 +142,21 @@ def test_data_reader_from_dataframe():
                        'falsies': ['inactive']},
         }
     }
-    df = data_reader(df, options=options)
+    df = data_reader(df, options=options, backend=backend)
 
-    print(df)
-    print(df.dtypes)
+    assert len(df) == 20
 
 
 @pytest.fixture(scope='module')
 def create_db(postgresql_proc):
-    db_credentials = {
-        'dbname': 'postgres',
-        'user': postgresql_proc.user,
-        'password': postgresql_proc.password,
-        'host': postgresql_proc.host,
-        'port': postgresql_proc.port,
-        'options': postgresql_proc.options
-    }
-
-    with psycopg.connect(**db_credentials) as db_connection:
+    uri = ('postgresql://{user}:{password}@{host}:{port}/{dbname}'
+           .format(dbname='postgres',
+                   user=postgresql_proc.user,
+                   password=postgresql_proc.password,
+                   host=postgresql_proc.host,
+                   port=postgresql_proc.port))
+
+    with psycopg2.connect(uri) as db_connection:
         with db_connection.cursor() as cursor:
             cursor.execute('''
                 CREATE SCHEMA deirokay;
@@ -176,16 +176,15 @@ def create_db(postgresql_proc):
             ''')
         db_connection.commit()
 
-    yield db_credentials
+    yield uri
 
-    with psycopg.connect(**db_credentials) as db_connection:
+    with psycopg2.connect(uri) as db_connection:
         with db_connection.cursor() as cursor:
             cursor.execute('DROP SCHEMA deirokay CASCADE;')
         db_connection.commit()
 
 
-def test_data_reader_from_sql_file(create_db):
-    db_credentials = create_db
+def test_data_reader_from_sql_file_pandas(create_db):  # pandas only
     options = {
         'columns': {
             'column1': {'dtype': 'string'},
@@ -195,16 +194,19 @@ def test_data_reader_from_sql_file(create_db):
             'column5': {'dtype': 'boolean'},
         }
     }
-    with psycopg.connect(**db_credentials) as con:
-        df = data_reader('tests/data_reader_from_sql_file.sql', options,
-                         con=con)
 
-    print(df)
-    print(df.dtypes)
+    df = data_reader('tests/data_reader_from_sql_file.sql', options,
+                     con=create_db,
+                     backend=Backend.PANDAS)
+
+    assert len(df) == 4
+
 
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_reader_from_sql_query(create_db, backend):
+    if backend == Backend.DASK:
+        pytest.xfail('Test for Dask backend not implemented')
 
-def test_data_reader_from_sql_query(create_db):
-    db_credentials = create_db
     options = {
         'columns': {
             'column1': {'dtype': 'string'},
@@ -214,9 +216,8 @@ def test_data_reader_from_sql_query(create_db):
             'column5': {'dtype': 'boolean'},
         }
     }
-    with psycopg.connect(**db_credentials) as con:
-        df = data_reader('select * from deirokay.test;', options,
-                         sql=True, con=con)
+    df = data_reader('select * from deirokay.test', options,
+                     sql=True, con=create_db,
+                     backend=backend)
 
-    print(df)
-    print(df.dtypes)
+    assert len(df) == 4
diff --git a/tests/test_data_treatment.py b/tests/test_data_treatment.py
index 78c1d40..a4d3dd0 100644
--- a/tests/test_data_treatment.py
+++ b/tests/test_data_treatment.py
@@ -4,10 +4,11 @@
 from numpy import nan
 from pandas.testing import assert_series_equal
 
-from deirokay.enums import DTypes
+from deirokay.enums import Backend, DTypes
 from deirokay.parser import get_dtype_treater, get_treater_instance
 
 
+@pytest.mark.parametrize('backend', list(Backend))
 @pytest.mark.parametrize('dtype, params, values', [
     (
         DTypes.INTEGER,
@@ -70,8 +71,8 @@
         ['on', 'off', None, True, False]
     )
 ])
-def test_dtype_parsing_for_Python_types(dtype, params, values):
-    treater_cls = get_dtype_treater(dtype)
+def test_dtype_parsing_for_Python_types(dtype, params, values, backend):
+    treater_cls = get_dtype_treater(dtype).attach_backend(backend)
     treater_instance = treater_cls(**params)
 
     # Use Deirokay to treat Python types
@@ -80,10 +81,16 @@ def test_dtype_parsing_for_Python_types(dtype, params, values):
     # Serialize with Deirokay
     serialized = json.dumps(treater_cls.serialize(parsed))
     json_parse = json.loads(serialized)
+    print(serialized)
 
     # Test for replication
-    parsed_from_serialized = (
-        get_treater_instance(json_parse['parser'])(json_parse['values'])
-    )
-    print(serialized)
+    parsed_from_serialized = get_treater_instance(
+        json_parse['parser'],
+        backend=backend
+    )(json_parse['values'])
+
+    if backend == Backend.DASK:
+        parsed = parsed.compute()
+        parsed_from_serialized = parsed_from_serialized.compute()
+
     assert_series_equal(parsed, parsed_from_serialized)
diff --git a/tests/test_data_validation.py b/tests/test_data_validation.py
index c6c6ed0..c25ad15 100644
--- a/tests/test_data_validation.py
+++ b/tests/test_data_validation.py
@@ -1,16 +1,18 @@
 import pytest
 
 from deirokay import data_reader, validate
-from deirokay.enums import SeverityLevel
+from deirokay.enums import Backend, SeverityLevel
 from deirokay.exceptions import ValidationError
 from deirokay.fs import split_s3_path
 
 
-def test_data_invalidation_from_dict():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_invalidation_from_dict(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -32,30 +34,36 @@ def test_data_invalidation_from_dict():
         validate(df, against=assertions)
 
 
-def test_data_validation_from_yaml():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_from_yaml(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.yaml'
+        options='tests/options.yaml',
+        backend=backend
     )
 
     validate(df, against='tests/assertions.yml')
 
 
-def test_data_validation_from_json():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_from_json(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     validate(df, against='tests/assertions.json')
 
 
-def test_custom_statement():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_custom_statement(backend):
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -91,10 +99,12 @@ def prepare_s3_custom_statement(require_s3_test_bucket):
     s3.delete_object(Bucket=bucket, Key=key)
 
 
-def test_custom_statement_from_s3(prepare_s3_custom_statement):
+@pytest.mark.parametrize('backend', list(Backend))
+def test_custom_statement_from_s3(prepare_s3_custom_statement, backend):
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -119,11 +129,13 @@ def test_custom_statement_from_s3(prepare_s3_custom_statement):
     validate(df, against=assertions)
 
 
-def test_data_validation_with_jinja():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_with_jinja(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -152,11 +164,13 @@ def test_data_validation_with_jinja():
     validate(df, against=assertions, template={'today': '20210816'})
 
 
-def test_data_validation_with_levels():
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_with_levels(backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
diff --git a/tests/test_history_validation.py b/tests/test_history_validation.py
index fc0bfce..13b104f 100644
--- a/tests/test_history_validation.py
+++ b/tests/test_history_validation.py
@@ -3,15 +3,18 @@
 import pytest
 
 from deirokay import data_reader, validate
+from deirokay.enums import Backend
 from deirokay.fs import fs_factory, split_s3_path
 from deirokay.history_template import series_from_fs
 
 
-def test_data_validation_with_jinja(prepare_history_folder):
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_with_jinja(prepare_history_folder, backend):
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
@@ -65,13 +68,16 @@ def delete_s3_prefix(bucket, prefix):
     delete_s3_prefix(bucket, prefix)
 
 
-def test_data_validation_with_jinja_using_s3(monkeypatch, prepare_history_s3):
+@pytest.mark.parametrize('backend', list(Backend))
+def test_data_validation_with_jinja_using_s3(monkeypatch, prepare_history_s3,
+                                             backend):
     # Reduce the number of retrieved objects to test pagination
     monkeypatch.setattr('deirokay.fs.S3FileSystem.LIST_OBJECTS_MAX_KEYS', 1)
 
     df = data_reader(
         'tests/transactions_sample.csv',
-        options='tests/options.json'
+        options='tests/options.json',
+        backend=backend
     )
 
     assertions = {
diff --git a/tests/test_multibackend_mixin.py b/tests/test_multibackend_mixin.py
new file mode 100644
index 0000000..898ff45
--- /dev/null
+++ b/tests/test_multibackend_mixin.py
@@ -0,0 +1,166 @@
+from typing import List
+
+import pytest
+
+from deirokay.backend import MultiBackendMixin, register_backend_method
+from deirokay.enums import Backend
+
+
+class BaseClassForTests(MultiBackendMixin):
+    def __init__(self, name) -> None:
+        self.name = name
+
+
+def test_non_multibackend_subclass():
+    with pytest.raises(RuntimeError):
+        class NotAMultibackendMiximSubclass():
+            @register_backend_method('report', Backend.PANDAS)
+            def _report(self):
+                ...
+
+
+class OneBackend(BaseClassForTests):
+    supported_backends: List[Backend] = [Backend.PANDAS]
+
+    @register_backend_method('report', Backend.PANDAS)
+    def _report(self):
+        return self.name, Backend.PANDAS
+
+    @register_backend_method('profile', Backend.PANDAS)
+    @staticmethod
+    def _profile():
+        return 'ok', Backend.PANDAS
+
+
+class TwoBackends(BaseClassForTests):
+    supported_backends: List[Backend] = [Backend.PANDAS, Backend.DASK]
+
+    @register_backend_method('report', Backend.DASK)
+    def _report_dask(self):
+        return self.name, Backend.DASK
+
+    @register_backend_method('report', Backend.PANDAS)
+    def _report_pandas(self):
+        return self.name, Backend.PANDAS
+
+    @register_backend_method('profile', Backend.DASK)
+    @staticmethod
+    def _profile_dask():
+        return 'ok', Backend.DASK
+
+    @register_backend_method('profile', Backend.PANDAS)
+    @staticmethod
+    def _profile_pandas():
+        return 'ok', Backend.PANDAS
+
+
+def test_use_valid_backend():
+    GeneratedClassA = OneBackend.attach_backend(Backend.PANDAS)
+    assert hasattr(GeneratedClassA, 'report')
+    assert GeneratedClassA('tvalue').report() == ('tvalue', Backend.PANDAS)
+    assert GeneratedClassA.profile() == ('ok', Backend.PANDAS)
+    assert GeneratedClassA('tvalue').profile() == ('ok', Backend.PANDAS)
+
+    GeneratedClassB1 = TwoBackends.attach_backend(Backend.PANDAS)
+    assert hasattr(GeneratedClassB1, 'report')
+    assert GeneratedClassB1('tvalue').report() == ('tvalue', Backend.PANDAS)
+    assert GeneratedClassB1.profile() == ('ok', Backend.PANDAS)
+    assert GeneratedClassB1('tvalue').profile() == ('ok', Backend.PANDAS)
+
+    GeneratedClassB2 = TwoBackends.attach_backend(Backend.DASK)
+    assert hasattr(GeneratedClassB2, 'report')
+    assert GeneratedClassB2('tvalue').report() == ('tvalue', Backend.DASK)
+    assert GeneratedClassB2.profile() == ('ok', Backend.DASK)
+    assert GeneratedClassB2('tvalue').profile() == ('ok', Backend.DASK)
+
+
+def test_use_unsupported_backend():
+    GeneratedClassA = OneBackend.attach_backend(Backend.PANDAS)
+    assert hasattr(GeneratedClassA, 'report')
+    with pytest.raises(AssertionError):
+        assert GeneratedClassA('tvalue').report() == ('tvalue', Backend.DASK)
+    with pytest.raises(AssertionError):
+        assert GeneratedClassA.profile() == ('ok', Backend.DASK)
+    with pytest.raises(AssertionError):
+        assert GeneratedClassA('tvalue').profile() == ('ok', Backend.DASK)
+
+
+def test_two_different_backends_in_a_row():
+    GeneratedClassA1 = TwoBackends.attach_backend(Backend.PANDAS)
+    GeneratedClassA2 = TwoBackends.attach_backend(Backend.DASK)
+
+    assert hasattr(GeneratedClassA1, 'report')
+    assert GeneratedClassA1('tvalue').report() == ('tvalue', Backend.PANDAS)
+    assert GeneratedClassA1.profile() == ('ok', Backend.PANDAS)
+    assert GeneratedClassA1('tvalue').profile() == ('ok', Backend.PANDAS)
+
+    assert hasattr(GeneratedClassA2, 'report')
+    assert GeneratedClassA2('tvalue').report() == ('tvalue', Backend.DASK)
+    assert GeneratedClassA2.profile() == ('ok', Backend.DASK)
+    assert GeneratedClassA2('tvalue').profile() == ('ok', Backend.DASK)
+
+
+class SimpleExtensionSubclass(TwoBackends):
+    ...
+
+
+def test_simple_subclass_of_multibackend_class():
+    GeneratedClassA = SimpleExtensionSubclass.attach_backend(Backend.DASK)
+    assert hasattr(GeneratedClassA, 'report')
+    assert GeneratedClassA('tvalue').report() == ('tvalue', Backend.DASK)
+    assert GeneratedClassA.profile() == ('ok', Backend.DASK)
+    assert GeneratedClassA('tvalue').profile() == ('ok', Backend.DASK)
+
+
+class ExtendAClassWithNewBackend(OneBackend):
+    supported_backends = OneBackend.supported_backends + [Backend.DASK]
+
+    @register_backend_method('report', Backend.DASK)
+    def _report_dask(self):
+        return self.name, Backend.DASK
+
+    @register_backend_method('profile', Backend.DASK)
+    @staticmethod
+    def _profile_dask():
+        return 'ok', Backend.DASK
+
+
+def test_subclass_extends_with_new_backend():
+    # Backend from parent
+    GeneratedClassA1 = ExtendAClassWithNewBackend.attach_backend(
+        Backend.PANDAS
+    )
+    assert hasattr(GeneratedClassA1, 'report')
+    assert GeneratedClassA1('tvalue').report() == ('tvalue', Backend.PANDAS)
+    assert GeneratedClassA1.profile() == ('ok', Backend.PANDAS)
+    assert GeneratedClassA1('tvalue').profile() == ('ok', Backend.PANDAS)
+
+    # New backend
+    GeneratedClassA2 = ExtendAClassWithNewBackend.attach_backend(
+        Backend.DASK
+    )
+    assert hasattr(GeneratedClassA2, 'report')
+    assert GeneratedClassA2('tvalue').report() == ('tvalue', Backend.DASK)
+    assert GeneratedClassA2.profile() == ('ok', Backend.DASK)
+    assert GeneratedClassA2('tvalue').profile() == ('ok', Backend.DASK)
+
+
+class ExtendAClassAndUseParentMethods(OneBackend):
+    @register_backend_method('report', Backend.PANDAS)
+    def _report(self):
+        return super()._report()
+
+    @register_backend_method('profile', Backend.PANDAS)
+    @staticmethod
+    def _profile():
+        return OneBackend._profile()
+
+
+def test_subclass_using_super():
+    GeneratedClassA = ExtendAClassAndUseParentMethods.attach_backend(
+        Backend.PANDAS
+    )
+    assert hasattr(GeneratedClassA, 'report')
+    assert GeneratedClassA('tvalue').report() == ('tvalue', Backend.PANDAS)
+    assert GeneratedClassA.profile() == ('ok', Backend.PANDAS)
+    assert GeneratedClassA('tvalue').profile() == ('ok', Backend.PANDAS)