register_spark_column_namespace

laktory.api.register_spark_column_namespace(name)

Decorator for registering a custom namespace on PySpark's Column class.

The decorated class receives a native PySpark Column as its first __init__ argument. Use this to build reusable column expressions that can be referenced from func_args strings when DATAFRAME_API=NATIVE.

.. note:: This namespace is NATIVE-only. A Narwhals Expr bridge is not provided because Narwhals expressions and PySpark Columns have incompatible semantics.

PARAMETER	DESCRIPTION
name	Namespace name, accessible as col("x").name.method() inside evaluated func_args strings. TYPE: str

Examples:

import laktory as lk


@lk.api.register_spark_column_namespace("custom")
class CustomColOps:
    def __init__(self, _col):
        self._col = _col

    def double(self):
        return self._col * 2

Use in a pipeline YAML:

transformer:
  nodes:
    - func_name: withColumn
      func_args:
        - x2
        - "col('x1').custom.double()"
      dataframe_api: NATIVE

References

• Spark Extension

Source code in laktory/api/namespace.py

def register_spark_column_namespace(name: str):
    """
    Decorator for registering a custom namespace on PySpark's ``Column`` class.

    The decorated class receives a **native PySpark Column** as its first
    ``__init__`` argument. Use this to build reusable column expressions that
    can be referenced from ``func_args`` strings when ``DATAFRAME_API=NATIVE``.

    .. note::
        This namespace is NATIVE-only. A Narwhals ``Expr`` bridge is not
        provided because Narwhals expressions and PySpark Columns have
        incompatible semantics.

    Parameters
    ----------
    name:
        Namespace name, accessible as ``col("x").name.method()`` inside
        evaluated ``func_args`` strings.

    Examples
    --------
    ```py
    import laktory as lk


    @lk.api.register_spark_column_namespace("custom")
    class CustomColOps:
        def __init__(self, _col):
            self._col = _col

        def double(self):
            return self._col * 2
    ```

    Use in a pipeline YAML:

    ```yaml
    transformer:
      nodes:
        - func_name: withColumn
          func_args:
            - x2
            - "col('x1').custom.double()"
          dataframe_api: NATIVE
    ```

    References
    ----------
    * [Spark Extension](https://www.laktory.ai/concepts/extension_custom/)
    """

    def wrapper(ns_cls: type):
        from pyspark.sql.column import Column

        setattr(Column, name, SparkNameSpace(name, ns_cls))
        try:
            from pyspark.sql.connect.column import Column as ConnectColumn

            setattr(ConnectColumn, name, SparkNameSpace(name, ns_cls))
        except ImportError:
            pass
        return ns_cls

    return wrapper