Google Git Google Git
Sign in
apache/datafusion-python/12aa11a0df345f28a64b60ea937bb4a5e3a7109e/./python/datafusion/functions/__init__.py
blob: 9fb20dc159b443fea7365861fc6ba96c2c44f167 [file]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""Scalar, aggregate, and window functions for :py:class:`~datafusion.expr.Expr`.
Each function returns an :py:class:`~datafusion.expr.Expr` that can be combined
with other expressions and passed to
:py:class:`~datafusion.dataframe.DataFrame` methods such as
:py:meth:`~datafusion.dataframe.DataFrame.select`,
:py:meth:`~datafusion.dataframe.DataFrame.filter`,
:py:meth:`~datafusion.dataframe.DataFrame.aggregate`, and
:py:meth:`~datafusion.dataframe.DataFrame.window`. The module is conventionally
imported as ``F`` so calls read like ``F.sum(col("price"))``.
Examples:
>>> from datafusion import functions as F
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3, 4]})
>>> df.aggregate([], [F.sum(col("a")).alias("total")]).to_pydict()
{'total': [10]}
See :ref:`aggregation` and :ref:`window_functions` in the online documentation
for categorized catalogs of aggregate and window functions.
"""
from __future__ import annotations
import inspect
import warnings
from typing import TYPE_CHECKING, Any
import pyarrow as pa
if TYPE_CHECKING:
from collections.abc import Callable, Iterable
from datafusion._internal import functions as f
from datafusion.common import NullTreatment
from datafusion.expr import (
CaseBuilder,
Expr,
SortExpr,
SortKey,
coerce_to_expr,
coerce_to_expr_list,
coerce_to_expr_or_none,
expr_list_to_raw_expr_list,
sort_list_to_raw_sort_list,
sort_or_default,
)
from datafusion.functions import spark
def _warn_if_expr_for_literal_arg(
value: Any, function_name: str, arg_name: str
) -> None:
if isinstance(value, Expr):
warnings.warn(
f"Passing Expr for {function_name}() argument {arg_name!r} is deprecated; "
"pass a Python literal instead.",
DeprecationWarning,
stacklevel=3,
)
__all__ = [
"abs",
"acos",
"acosh",
"alias",
"any_match",
"approx_distinct",
"approx_median",
"approx_percentile_cont",
"approx_percentile_cont_with_weight",
"array",
"array_agg",
"array_any_match",
"array_any_value",
"array_append",
"array_cat",
"array_compact",
"array_concat",
"array_contains",
"array_dims",
"array_distance",
"array_distinct",
"array_element",
"array_empty",
"array_except",
"array_extract",
"array_filter",
"array_has",
"array_has_all",
"array_has_any",
"array_indexof",
"array_intersect",
"array_join",
"array_length",
"array_max",
"array_min",
"array_ndims",
"array_normalize",
"array_pop_back",
"array_pop_front",
"array_position",
"array_positions",
"array_prepend",
"array_push_back",
"array_push_front",
"array_remove",
"array_remove_all",
"array_remove_n",
"array_repeat",
"array_replace",
"array_replace_all",
"array_replace_n",
"array_resize",
"array_reverse",
"array_slice",
"array_sort",
"array_to_string",
"array_transform",
"array_union",
"arrays_overlap",
"arrays_zip",
"arrow_cast",
"arrow_field",
"arrow_metadata",
"arrow_try_cast",
"arrow_typeof",
"ascii",
"asin",
"asinh",
"atan",
"atan2",
"atanh",
"avg",
"bit_and",
"bit_length",
"bit_or",
"bit_xor",
"bool_and",
"bool_or",
"btrim",
"cardinality",
"case",
"cast_to_type",
"cbrt",
"ceil",
"char_length",
"character_length",
"chr",
"coalesce",
"col",
"concat",
"concat_ws",
"contains",
"corr",
"cos",
"cosh",
"cosine_distance",
"cot",
"count",
"count_star",
"covar",
"covar_pop",
"covar_samp",
"cume_dist",
"current_date",
"current_time",
"current_timestamp",
"date_bin",
"date_format",
"date_part",
"date_trunc",
"datepart",
"datetrunc",
"decode",
"degrees",
"dense_rank",
"digest",
"dot_product",
"element_at",
"empty",
"encode",
"ends_with",
"exp",
"extract",
"factorial",
"find_in_set",
"first_value",
"flatten",
"floor",
"from_unixtime",
"gcd",
"gen_series",
"generate_series",
"get_field",
"greatest",
"grouping",
"ifnull",
"in_list",
"initcap",
"inner_product",
"instr",
"isnan",
"iszero",
"lag",
"lambda_",
"lambda_var",
"last_value",
"lcm",
"lead",
"least",
"left",
"length",
"levenshtein",
"list_any_match",
"list_any_value",
"list_append",
"list_cat",
"list_compact",
"list_concat",
"list_contains",
"list_dims",
"list_distance",
"list_distinct",
"list_element",
"list_empty",
"list_except",
"list_extract",
"list_filter",
"list_has",
"list_has_all",
"list_has_any",
"list_indexof",
"list_intersect",
"list_join",
"list_length",
"list_max",
"list_min",
"list_ndims",
"list_normalize",
"list_overlap",
"list_pop_back",
"list_pop_front",
"list_position",
"list_positions",
"list_prepend",
"list_push_back",
"list_push_front",
"list_remove",
"list_remove_all",
"list_remove_n",
"list_repeat",
"list_replace",
"list_replace_all",
"list_replace_n",
"list_resize",
"list_reverse",
"list_slice",
"list_sort",
"list_to_string",
"list_transform",
"list_union",
"list_zip",
"ln",
"log",
"log2",
"log10",
"lower",
"lpad",
"ltrim",
"make_array",
"make_date",
"make_list",
"make_map",
"make_time",
"map_entries",
"map_extract",
"map_keys",
"map_values",
"max",
"md5",
"mean",
"median",
"min",
"named_struct",
"nanvl",
"now",
"nth_value",
"ntile",
"nullif",
"nvl",
"nvl2",
"octet_length",
"order_by",
"overlay",
"percent_rank",
"percentile_cont",
"pi",
"position",
"pow",
"power",
"quantile_cont",
"radians",
"random",
"range",
"rank",
"regexp_count",
"regexp_instr",
"regexp_like",
"regexp_match",
"regexp_replace",
"regr_avgx",
"regr_avgy",
"regr_count",
"regr_intercept",
"regr_r2",
"regr_slope",
"regr_sxx",
"regr_sxy",
"regr_syy",
"repeat",
"replace",
"reverse",
"right",
"round",
"row",
"row_number",
"rpad",
"rtrim",
"sha224",
"sha256",
"sha384",
"sha512",
"signum",
"sin",
"sinh",
"spark",
"split_part",
"sqrt",
"starts_with",
"stddev",
"stddev_pop",
"stddev_samp",
"string_agg",
"string_to_array",
"string_to_list",
"strpos",
"struct",
"substr",
"substr_index",
"substring",
"sum",
"tan",
"tanh",
"to_char",
"to_date",
"to_hex",
"to_local_time",
"to_time",
"to_timestamp",
"to_timestamp_micros",
"to_timestamp_millis",
"to_timestamp_nanos",
"to_timestamp_seconds",
"to_unixtime",
"today",
"translate",
"trim",
"trunc",
"try_cast_to_type",
"union_extract",
"union_tag",
"upper",
"uuid",
"var",
"var_pop",
"var_population",
"var_samp",
"var_sample",
"version",
"when",
"with_metadata",
]
def isnan(expr: Expr) -> Expr:
"""Returns true if a given number is +NaN or -NaN otherwise returns false.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, np.nan]})
>>> result = df.select(dfn.functions.isnan(dfn.col("a")).alias("isnan"))
>>> result.collect_column("isnan")[1].as_py()
True
"""
return Expr(f.isnan(expr.expr))
def nullif(expr1: Expr, expr2: Expr) -> Expr:
"""Returns NULL if expr1 equals expr2; otherwise it returns expr1.
This can be used to perform the inverse operation of the COALESCE expression.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2], "b": [1, 3]})
>>> result = df.select(
... dfn.functions.nullif(dfn.col("a"), dfn.col("b")).alias("nullif"))
>>> result.collect_column("nullif").to_pylist()
[None, 2]
"""
return Expr(f.nullif(expr1.expr, expr2.expr))
def encode(expr: Expr, encoding: Expr | str) -> Expr:
"""Encode the ``input``, using the ``encoding``. encoding can be base64 or hex.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.encode(dfn.col("a"), "base64").alias("enc"))
>>> result.collect_column("enc")[0].as_py()
'aGVsbG8'
"""
_warn_if_expr_for_literal_arg(encoding, "encode", "encoding")
encoding = coerce_to_expr(encoding)
return Expr(f.encode(expr.expr, encoding.expr))
def decode(expr: Expr, encoding: Expr | str) -> Expr:
"""Decode the ``input``, using the ``encoding``. encoding can be base64 or hex.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["aGVsbG8="]})
>>> result = df.select(
... dfn.functions.decode(dfn.col("a"), "base64").alias("dec"))
>>> result.collect_column("dec")[0].as_py()
b'hello'
"""
_warn_if_expr_for_literal_arg(encoding, "decode", "encoding")
encoding = coerce_to_expr(encoding)
return Expr(f.decode(expr.expr, encoding.expr))
def array_to_string(expr: Expr, delimiter: Expr | str) -> Expr:
"""Converts each element to its text representation.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_to_string(dfn.col("a"), ",").alias("s"))
>>> result.collect_column("s")[0].as_py()
'1,2,3'
"""
delimiter = coerce_to_expr(delimiter)
return Expr(f.array_to_string(expr.expr, delimiter.expr.cast(pa.string())))
def array_join(expr: Expr, delimiter: Expr | str) -> Expr:
"""Converts each element to its text representation.
See Also:
This is an alias for :py:func:`array_to_string`.
"""
return array_to_string(expr, delimiter)
def list_to_string(expr: Expr, delimiter: Expr | str) -> Expr:
"""Converts each element to its text representation.
See Also:
This is an alias for :py:func:`array_to_string`.
"""
return array_to_string(expr, delimiter)
def list_join(expr: Expr, delimiter: Expr | str) -> Expr:
"""Converts each element to its text representation.
See Also:
This is an alias for :py:func:`array_to_string`.
"""
return array_to_string(expr, delimiter)
def lambda_var(name: str) -> Expr:
"""Create an unresolved reference to a lambda parameter by ``name``.
Use this inside the body passed to :py:func:`lambda_` to refer to one of the
lambda's parameters. The owning higher-order function (such as
:py:func:`array_transform`) binds the variable to a concrete element type
during query planning.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> double_fn = F.lambda_(["v"], F.lambda_var("v") * lit(2))
>>> df.select(
... F.array_transform(col("a"), double_fn).alias("d")
... ).collect_column("d")[0].as_py()
[2, 4, 6]
See Also:
:py:func:`lambda_`, :py:func:`array_transform`, :py:func:`array_any_match`.
"""
return Expr(f.lambda_var(name))
def lambda_(params: list[str], body: Expr) -> Expr:
"""Create a lambda expression from parameter names and a body expression.
This is the explicit form of building a lambda. Most callers can instead
pass a Python callable directly to a higher-order function such as
:py:func:`array_transform`, which builds the lambda automatically. Reach for
``lambda_`` when you want explicit control over the parameter names.
Args:
params: Ordered lambda parameter names.
body: Body expression that references the parameters via
:py:func:`lambda_var`.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> double_fn = F.lambda_(["v"], F.lambda_var("v") * lit(2))
>>> df.select(
... F.array_transform(col("a"), double_fn).alias("d")
... ).collect_column("d")[0].as_py()
[2, 4, 6]
See Also:
:py:func:`lambda_var`, :py:func:`array_transform`, :py:func:`array_any_match`.
"""
return Expr(f.lambda_(params, body.expr))
def _to_lambda(fn: Expr | Callable[..., Any]) -> Expr:
"""Coerce ``fn`` to a lambda ``Expr``.
Accepts either an ``Expr`` produced by :py:func:`lambda_` (returned
unchanged) or a Python callable. A callable is introspected for its
parameter names; those names become :py:func:`lambda_var` references passed
positionally into the callable, and its return value (coerced to an
``Expr``) becomes the lambda body.
"""
if isinstance(fn, Expr):
return fn
if not callable(fn):
msg = f"expected an Expr or callable, got {type(fn).__name__}"
raise TypeError(msg)
params = list(inspect.signature(fn).parameters)
if not params:
msg = "lambda callable must accept at least one parameter"
raise ValueError(msg)
body = coerce_to_expr(fn(*[lambda_var(p) for p in params]))
return lambda_(params, body)
def array_transform(array: Expr, transform: Expr | Callable[..., Any]) -> Expr:
"""Transform each element of ``array`` with a lambda.
``transform`` may be a Python callable, which is converted to a lambda
automatically (its parameter names become the lambda parameters), or an
explicit lambda built with :py:func:`lambda_`.
Examples:
Using a Python callable:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> df.select(
... F.array_transform(col("a"), lambda v: v * 2).alias("d")
... ).collect_column("d")[0].as_py()
[2, 4, 6]
Using an explicit lambda built with :py:func:`lambda_`:
>>> double_fn = F.lambda_(["v"], F.lambda_var("v") * lit(2))
>>> df.select(
... F.array_transform(col("a"), double_fn).alias("d")
... ).collect_column("d")[0].as_py()
[2, 4, 6]
See Also:
:py:func:`array_any_match`, :py:func:`lambda_`.
"""
return Expr(f.array_transform(array.expr, _to_lambda(transform).expr))
def list_transform(array: Expr, transform: Expr | Callable[..., Any]) -> Expr:
"""Transform each element of a list with a lambda.
See Also:
This is an alias for :py:func:`array_transform`.
"""
return array_transform(array, transform)
def array_any_match(array: Expr, predicate: Expr | Callable[..., Any]) -> Expr:
"""Return ``True`` if any element of ``array`` satisfies ``predicate``.
``predicate`` may be a Python callable, converted to a lambda
automatically, or an explicit lambda built with :py:func:`lambda_`. It must
return a boolean expression.
Examples:
Using a Python callable:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> df.select(
... F.array_any_match(col("a"), lambda v: v > 2).alias("m")
... ).collect_column("m")[0].as_py()
True
Using an explicit lambda built with :py:func:`lambda_`:
>>> predicate = F.lambda_(["v"], F.lambda_var("v") > lit(2))
>>> df.select(
... F.array_any_match(col("a"), predicate).alias("m")
... ).collect_column("m")[0].as_py()
True
See Also:
:py:func:`array_transform`, :py:func:`lambda_`.
"""
return Expr(f.array_any_match(array.expr, _to_lambda(predicate).expr))
def any_match(array: Expr, predicate: Expr | Callable[..., Any]) -> Expr:
"""Return ``True`` if any element of an array satisfies a predicate.
See Also:
This is an alias for :py:func:`array_any_match`.
"""
return array_any_match(array, predicate)
def list_any_match(array: Expr, predicate: Expr | Callable[..., Any]) -> Expr:
"""Return ``True`` if any element of a list satisfies a predicate.
See Also:
This is an alias for :py:func:`array_any_match`.
"""
return array_any_match(array, predicate)
def array_filter(array: Expr, predicate: Expr | Callable[..., Any]) -> Expr:
"""Keep the elements of ``array`` for which ``predicate`` is ``True``.
``predicate`` may be a Python callable, converted to a lambda
automatically, or an explicit lambda built with :py:func:`lambda_`. It must
return a boolean expression. The result is a new array containing only the
matching elements.
Examples:
Using a Python callable:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3, 4, 5]]})
>>> df.select(
... F.array_filter(col("a"), lambda v: v > 2).alias("f")
... ).collect_column("f")[0].as_py()
[3, 4, 5]
Using an explicit lambda built with :py:func:`lambda_`:
>>> predicate = F.lambda_(["v"], F.lambda_var("v") > lit(2))
>>> df.select(
... F.array_filter(col("a"), predicate).alias("f")
... ).collect_column("f")[0].as_py()
[3, 4, 5]
See Also:
:py:func:`array_transform`, :py:func:`array_any_match`, :py:func:`lambda_`.
"""
return Expr(f.array_filter(array.expr, _to_lambda(predicate).expr))
def list_filter(array: Expr, predicate: Expr | Callable[..., Any]) -> Expr:
"""Keep the elements of a list for which a predicate is ``True``.
See Also:
This is an alias for :py:func:`array_filter`.
"""
return array_filter(array, predicate)
def in_list(arg: Expr, values: list[Expr], negated: bool = False) -> Expr:
"""Returns whether the argument is contained within the list ``values``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.select(
... dfn.functions.in_list(
... dfn.col("a"), [dfn.lit(1), dfn.lit(3)]
... ).alias("in")
... )
>>> result.collect_column("in").to_pylist()
[True, False, True]
>>> result = df.select(
... dfn.functions.in_list(
... dfn.col("a"), [dfn.lit(1), dfn.lit(3)],
... negated=True,
... ).alias("not_in")
... )
>>> result.collect_column("not_in").to_pylist()
[False, True, False]
"""
values = [v.expr for v in values]
return Expr(f.in_list(arg.expr, values, negated))
def digest(value: Expr, method: Expr | str) -> Expr:
"""Computes the binary hash of an expression using the specified algorithm.
Standard algorithms are md5, sha224, sha256, sha384, sha512, blake2s,
blake2b, and blake3.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.digest(dfn.col("a"), "md5").alias("d"))
>>> len(result.collect_column("d")[0].as_py()) > 0
True
"""
_warn_if_expr_for_literal_arg(method, "digest", "method")
method = coerce_to_expr(method)
return Expr(f.digest(value.expr, method.expr))
def contains(string: Expr, search_str: Expr | str) -> Expr:
"""Returns true if ``search_str`` is found within ``string`` (case-sensitive).
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["the quick brown fox"]})
>>> result = df.select(
... dfn.functions.contains(dfn.col("a"), "brown").alias("c"))
>>> result.collect_column("c")[0].as_py()
True
"""
search_str = coerce_to_expr(search_str)
return Expr(f.contains(string.expr, search_str.expr))
def concat(*args: Expr) -> Expr:
"""Concatenates the text representations of all the arguments.
NULL arguments are ignored.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"], "b": [" world"]})
>>> result = df.select(
... dfn.functions.concat(dfn.col("a"), dfn.col("b")).alias("c")
... )
>>> result.collect_column("c")[0].as_py()
'hello world'
"""
args = [arg.expr for arg in args]
return Expr(f.concat(args))
def concat_ws(separator: str, *args: Expr) -> Expr:
"""Concatenates the list ``args`` with the separator.
``NULL`` arguments are ignored. ``separator`` should not be ``NULL``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"], "b": ["world"]})
>>> result = df.select(
... dfn.functions.concat_ws("-", dfn.col("a"), dfn.col("b")).alias("c"))
>>> result.collect_column("c")[0].as_py()
'hello-world'
"""
args = [arg.expr for arg in args]
return Expr(f.concat_ws(separator, args))
def order_by(expr: Expr, ascending: bool = True, nulls_first: bool = True) -> SortExpr:
"""Creates a new sort expression.
Examples:
>>> sort_expr = dfn.functions.order_by(
... dfn.col("a"), ascending=False)
>>> sort_expr.ascending()
False
>>> sort_expr = dfn.functions.order_by(
... dfn.col("a"), ascending=True, nulls_first=False)
>>> sort_expr.nulls_first()
False
"""
return SortExpr(expr, ascending=ascending, nulls_first=nulls_first)
def alias(expr: Expr, name: str, metadata: dict[str, str] | None = None) -> Expr:
"""Creates an alias expression with an optional metadata dictionary.
Args:
expr: The expression to alias
name: The alias name
metadata: Optional metadata to attach to the column
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2]})
>>> result = df.select(
... dfn.functions.alias(
... dfn.col("a"), "b"
... )
... )
>>> result.collect_column("b")[0].as_py()
1
>>> result = df.select(
... dfn.functions.alias(
... dfn.col("a"), "b", metadata={"info": "test"}
... )
... )
>>> result.schema()
b: int64
-- field metadata --
info: 'test'
"""
return Expr(f.alias(expr.expr, name, metadata))
def col(name: str) -> Expr:
"""Creates a column reference expression.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> df.select(dfn.functions.col("a")).collect_column("a")[0].as_py()
1
"""
return Expr(f.col(name))
def count_star(filter: Expr | None = None) -> Expr:
"""Create a COUNT(1) aggregate expression.
This aggregate function will count all of the rows in the partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``distinct``, and ``null_treatment``.
Args:
filter: If provided, only count rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.count_star(
... ).alias("cnt")])
>>> result.collect_column("cnt")[0].as_py()
3
>>> result = df.aggregate(
... [], [dfn.functions.count_star(
... filter=dfn.col("a") > dfn.lit(1)
... ).alias("cnt")])
>>> result.collect_column("cnt")[0].as_py()
2
"""
return count(Expr.literal(1), filter=filter)
def case(expr: Expr) -> CaseBuilder:
"""Create a case expression.
Create a :py:class:`~datafusion.expr.CaseBuilder` to match cases for the
expression ``expr``. See :py:class:`~datafusion.expr.CaseBuilder` for
detailed usage.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.select(
... dfn.functions.case(dfn.col("a")).when(dfn.lit(1),
... dfn.lit("one")).otherwise(dfn.lit("other")).alias("c"))
>>> result.collect_column("c")[0].as_py()
'one'
"""
return CaseBuilder(f.case(expr.expr))
def when(when: Expr, then: Expr) -> CaseBuilder:
"""Create a case expression that has no base expression.
Create a :py:class:`~datafusion.expr.CaseBuilder` to match cases for the
expression ``expr``. See :py:class:`~datafusion.expr.CaseBuilder` for
detailed usage.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.select(
... dfn.functions.when(dfn.col("a") > dfn.lit(2),
... dfn.lit("big")).otherwise(dfn.lit("small")).alias("c"))
>>> result.collect_column("c")[2].as_py()
'big'
"""
return CaseBuilder(f.when(when.expr, then.expr))
# scalar functions
def abs(arg: Expr) -> Expr:
"""Return the absolute value of a given number.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [-1, 0, 1]})
>>> result = df.select(dfn.functions.abs(dfn.col("a")).alias("abs"))
>>> result.collect_column("abs")[0].as_py()
1
"""
return Expr(f.abs(arg.expr))
def acos(arg: Expr) -> Expr:
"""Returns the arc cosine or inverse cosine of a number.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0]})
>>> result = df.select(dfn.functions.acos(dfn.col("a")).alias("acos"))
>>> result.collect_column("acos")[0].as_py()
0.0
"""
return Expr(f.acos(arg.expr))
def acosh(arg: Expr) -> Expr:
"""Returns inverse hyperbolic cosine.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0]})
>>> result = df.select(dfn.functions.acosh(dfn.col("a")).alias("acosh"))
>>> result.collect_column("acosh")[0].as_py()
0.0
"""
return Expr(f.acosh(arg.expr))
def ascii(arg: Expr) -> Expr:
"""Returns the numeric code of the first character of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["a","b","c"]})
>>> ascii_df = df.select(dfn.functions.ascii(dfn.col("a")).alias("ascii"))
>>> ascii_df.collect_column("ascii")[0].as_py()
97
"""
return Expr(f.ascii(arg.expr))
def asin(arg: Expr) -> Expr:
"""Returns the arc sine or inverse sine of a number.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.asin(dfn.col("a")).alias("asin"))
>>> result.collect_column("asin")[0].as_py()
0.0
"""
return Expr(f.asin(arg.expr))
def asinh(arg: Expr) -> Expr:
"""Returns inverse hyperbolic sine.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.asinh(dfn.col("a")).alias("asinh"))
>>> result.collect_column("asinh")[0].as_py()
0.0
"""
return Expr(f.asinh(arg.expr))
def atan(arg: Expr) -> Expr:
"""Returns inverse tangent of a number.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.atan(dfn.col("a")).alias("atan"))
>>> result.collect_column("atan")[0].as_py()
0.0
"""
return Expr(f.atan(arg.expr))
def atanh(arg: Expr) -> Expr:
"""Returns inverse hyperbolic tangent.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.atanh(dfn.col("a")).alias("atanh"))
>>> result.collect_column("atanh")[0].as_py()
0.0
"""
return Expr(f.atanh(arg.expr))
def atan2(y: Expr, x: Expr) -> Expr:
"""Returns inverse tangent of a division given in the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [0.0], "x": [1.0]})
>>> result = df.select(
... dfn.functions.atan2(dfn.col("y"), dfn.col("x")).alias("atan2"))
>>> result.collect_column("atan2")[0].as_py()
0.0
"""
return Expr(f.atan2(y.expr, x.expr))
def bit_length(arg: Expr) -> Expr:
"""Returns the number of bits in the string argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["a","b","c"]})
>>> bit_df = df.select(dfn.functions.bit_length(dfn.col("a")).alias("bit_len"))
>>> bit_df.collect_column("bit_len")[0].as_py()
8
"""
return Expr(f.bit_length(arg.expr))
def btrim(arg: Expr) -> Expr:
"""Removes all characters, spaces by default, from both sides of a string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [" a "]})
>>> trim_df = df.select(dfn.functions.btrim(dfn.col("a")).alias("trimmed"))
>>> trim_df.collect_column("trimmed")[0].as_py()
'a'
"""
return Expr(f.btrim(arg.expr))
def cbrt(arg: Expr) -> Expr:
"""Returns the cube root of a number.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [27]})
>>> cbrt_df = df.select(dfn.functions.cbrt(dfn.col("a")).alias("cbrt"))
>>> cbrt_df.collect_column("cbrt")[0].as_py()
3.0
"""
return Expr(f.cbrt(arg.expr))
def ceil(arg: Expr) -> Expr:
"""Returns the nearest integer greater than or equal to argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.9]})
>>> ceil_df = df.select(dfn.functions.ceil(dfn.col("a")).alias("ceil"))
>>> ceil_df.collect_column("ceil")[0].as_py()
2.0
"""
return Expr(f.ceil(arg.expr))
def character_length(arg: Expr) -> Expr:
"""Returns the number of characters in the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["abc","b","c"]})
>>> char_len_df = df.select(
... dfn.functions.character_length(dfn.col("a")).alias("char_len"))
>>> char_len_df.collect_column("char_len")[0].as_py()
3
"""
return Expr(f.character_length(arg.expr))
def length(string: Expr) -> Expr:
"""The number of characters in the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.length(dfn.col("a")).alias("len"))
>>> result.collect_column("len")[0].as_py()
5
"""
return Expr(f.length(string.expr))
def char_length(string: Expr) -> Expr:
"""The number of characters in the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.char_length(dfn.col("a")).alias("len"))
>>> result.collect_column("len")[0].as_py()
5
"""
return Expr(f.char_length(string.expr))
def chr(arg: Expr) -> Expr:
"""Converts the Unicode code point to a UTF8 character.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [65]})
>>> result = df.select(dfn.functions.chr(dfn.col("a")).alias("chr"))
>>> result.collect_column("chr")[0].as_py()
'A'
"""
return Expr(f.chr(arg.expr))
def coalesce(*args: Expr) -> Expr:
"""Returns the value of the first expr in ``args`` which is not NULL.
Args:
*args: Expressions to evaluate in order.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [None, 1], "b": [2, 3]})
>>> result = df.select(
... dfn.functions.coalesce(dfn.col("a"), dfn.col("b")).alias("c"))
>>> result.collect_column("c")[0].as_py()
2
"""
args = [arg.expr for arg in args]
return Expr(f.coalesce(*args))
def cos(arg: Expr) -> Expr:
"""Returns the cosine of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0,-1,1]})
>>> cos_df = df.select(dfn.functions.cos(dfn.col("a")).alias("cos"))
>>> cos_df.collect_column("cos")[0].as_py()
1.0
"""
return Expr(f.cos(arg.expr))
def cosh(arg: Expr) -> Expr:
"""Returns the hyperbolic cosine of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0,-1,1]})
>>> cosh_df = df.select(dfn.functions.cosh(dfn.col("a")).alias("cosh"))
>>> cosh_df.collect_column("cosh")[0].as_py()
1.0
"""
return Expr(f.cosh(arg.expr))
def cot(arg: Expr) -> Expr:
"""Returns the cotangent of the argument.
Examples:
>>> from math import pi
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [pi / 4]})
>>> result = df.select(
... dfn.functions.cot(dfn.col("a")).alias("cot")
... )
>>> result.collect_column("cot")[0].as_py()
1.0...
"""
return Expr(f.cot(arg.expr))
def degrees(arg: Expr) -> Expr:
"""Converts the argument from radians to degrees.
Examples:
>>> from math import pi
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0,pi,2*pi]})
>>> deg_df = df.select(dfn.functions.degrees(dfn.col("a")).alias("deg"))
>>> deg_df.collect_column("deg")[2].as_py()
360.0
"""
return Expr(f.degrees(arg.expr))
def ends_with(arg: Expr, suffix: Expr | str) -> Expr:
"""Returns true if the ``string`` ends with the ``suffix``, false otherwise.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["abc","b","c"]})
>>> ends_with_df = df.select(
... dfn.functions.ends_with(dfn.col("a"), "c").alias("ends_with"))
>>> ends_with_df.collect_column("ends_with")[0].as_py()
True
"""
suffix = coerce_to_expr(suffix)
return Expr(f.ends_with(arg.expr, suffix.expr))
def exp(arg: Expr) -> Expr:
"""Returns the exponential of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.exp(dfn.col("a")).alias("exp"))
>>> result.collect_column("exp")[0].as_py()
1.0
"""
return Expr(f.exp(arg.expr))
def factorial(arg: Expr) -> Expr:
"""Returns the factorial of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [3]})
>>> result = df.select(
... dfn.functions.factorial(dfn.col("a")).alias("factorial")
... )
>>> result.collect_column("factorial")[0].as_py()
6
"""
return Expr(f.factorial(arg.expr))
def find_in_set(string: Expr, string_list: Expr | str) -> Expr:
"""Find a string in a list of strings.
Returns a value in the range of 1 to N if the string is in the string list
``string_list`` consisting of N substrings.
The string list is a string composed of substrings separated by ``,`` characters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["b"]})
>>> result = df.select(
... dfn.functions.find_in_set(dfn.col("a"), "a,b,c").alias("pos"))
>>> result.collect_column("pos")[0].as_py()
2
"""
string_list = coerce_to_expr(string_list)
return Expr(f.find_in_set(string.expr, string_list.expr))
def floor(arg: Expr) -> Expr:
"""Returns the nearest integer less than or equal to the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.9]})
>>> floor_df = df.select(dfn.functions.floor(dfn.col("a")).alias("floor"))
>>> floor_df.collect_column("floor")[0].as_py()
1.0
"""
return Expr(f.floor(arg.expr))
def gcd(x: Expr, y: Expr) -> Expr:
"""Returns the greatest common divisor.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [12], "b": [8]})
>>> result = df.select(
... dfn.functions.gcd(dfn.col("a"), dfn.col("b")).alias("gcd")
... )
>>> result.collect_column("gcd")[0].as_py()
4
"""
return Expr(f.gcd(x.expr, y.expr))
def greatest(*args: Expr) -> Expr:
"""Returns the greatest value from a list of expressions.
Returns NULL if all expressions are NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 3], "b": [2, 1]})
>>> result = df.select(
... dfn.functions.greatest(dfn.col("a"), dfn.col("b")).alias("greatest"))
>>> result.collect_column("greatest")[0].as_py()
2
>>> result.collect_column("greatest")[1].as_py()
3
"""
exprs = [arg.expr for arg in args]
return Expr(f.greatest(*exprs))
def ifnull(x: Expr, y: Expr) -> Expr:
"""Returns ``x`` if ``x`` is not NULL. Otherwise returns ``y``.
Args:
x: Expression to return when it is not NULL.
y: Fallback expression to return when ``x`` is NULL.
See Also:
This is an alias for :py:func:`nvl`.
"""
return nvl(x, y)
def initcap(string: Expr) -> Expr:
"""Set the initial letter of each word to capital.
Converts the first letter of each word in ``string`` to uppercase and the remaining
characters to lowercase.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["the cat"]})
>>> cap_df = df.select(dfn.functions.initcap(dfn.col("a")).alias("cap"))
>>> cap_df.collect_column("cap")[0].as_py()
'The Cat'
"""
return Expr(f.initcap(string.expr))
def instr(string: Expr, substring: Expr | str) -> Expr:
"""Finds the position from where the ``substring`` matches the ``string``.
See Also:
This is an alias for :py:func:`strpos`.
"""
return strpos(string, substring)
def iszero(arg: Expr) -> Expr:
"""Returns true if a given number is +0.0 or -0.0 otherwise returns false.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0, 1.0]})
>>> result = df.select(dfn.functions.iszero(dfn.col("a")).alias("iz"))
>>> result.collect_column("iz")[0].as_py()
True
"""
return Expr(f.iszero(arg.expr))
def lcm(x: Expr, y: Expr) -> Expr:
"""Returns the least common multiple.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [4], "b": [6]})
>>> result = df.select(
... dfn.functions.lcm(dfn.col("a"), dfn.col("b")).alias("lcm")
... )
>>> result.collect_column("lcm")[0].as_py()
12
"""
return Expr(f.lcm(x.expr, y.expr))
def least(*args: Expr) -> Expr:
"""Returns the least value from a list of expressions.
Returns NULL if all expressions are NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 3], "b": [2, 1]})
>>> result = df.select(
... dfn.functions.least(dfn.col("a"), dfn.col("b")).alias("least"))
>>> result.collect_column("least")[0].as_py()
1
>>> result.collect_column("least")[1].as_py()
1
"""
exprs = [arg.expr for arg in args]
return Expr(f.least(*exprs))
def left(string: Expr, n: Expr | int) -> Expr:
"""Returns the first ``n`` characters in the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["the cat"]})
>>> left_df = df.select(
... dfn.functions.left(dfn.col("a"), 3).alias("left"))
>>> left_df.collect_column("left")[0].as_py()
'the'
"""
n = coerce_to_expr(n)
return Expr(f.left(string.expr, n.expr))
def levenshtein(string1: Expr, string2: Expr | str) -> Expr:
"""Returns the Levenshtein distance between the two given strings.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["kitten"]})
>>> result = df.select(
... dfn.functions.levenshtein(dfn.col("a"), "sitting").alias("d"))
>>> result.collect_column("d")[0].as_py()
3
"""
string2 = coerce_to_expr(string2)
return Expr(f.levenshtein(string1.expr, string2.expr))
def ln(arg: Expr) -> Expr:
"""Returns the natural logarithm (base e) of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0]})
>>> result = df.select(dfn.functions.ln(dfn.col("a")).alias("ln"))
>>> result.collect_column("ln")[0].as_py()
0.0
"""
return Expr(f.ln(arg.expr))
def log(base: Expr | int | float, num: Expr) -> Expr: # noqa: PYI041
"""Returns the logarithm of a number for a particular ``base``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [100.0]})
>>> result = df.select(
... dfn.functions.log(10.0, dfn.col("a")).alias("log")
... )
>>> result.collect_column("log")[0].as_py()
2.0
"""
base = coerce_to_expr(base)
return Expr(f.log(base.expr, num.expr))
def log10(arg: Expr) -> Expr:
"""Base 10 logarithm of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [100.0]})
>>> result = df.select(dfn.functions.log10(dfn.col("a")).alias("log10"))
>>> result.collect_column("log10")[0].as_py()
2.0
"""
return Expr(f.log10(arg.expr))
def log2(arg: Expr) -> Expr:
"""Base 2 logarithm of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [8.0]})
>>> result = df.select(dfn.functions.log2(dfn.col("a")).alias("log2"))
>>> result.collect_column("log2")[0].as_py()
3.0
"""
return Expr(f.log2(arg.expr))
def lower(arg: Expr) -> Expr:
"""Converts a string to lowercase.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["THE CaT"]})
>>> lower_df = df.select(dfn.functions.lower(dfn.col("a")).alias("lower"))
>>> lower_df.collect_column("lower")[0].as_py()
'the cat'
"""
return Expr(f.lower(arg.expr))
def lpad(string: Expr, count: Expr | int, characters: Expr | str | None = None) -> Expr:
"""Add left padding to a string.
Extends the string to length length by prepending the characters fill (a
space by default). If the string is already longer than length then it is
truncated (on the right).
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["the cat", "a hat"]})
>>> lpad_df = df.select(
... dfn.functions.lpad(dfn.col("a"), 6).alias("lpad"))
>>> lpad_df.collect_column("lpad")[0].as_py()
'the ca'
>>> lpad_df.collect_column("lpad")[1].as_py()
' a hat'
>>> result = df.select(
... dfn.functions.lpad(
... dfn.col("a"), 10, characters="."
... ).alias("lpad"))
>>> result.collect_column("lpad")[0].as_py()
'...the cat'
"""
count = coerce_to_expr(count)
characters = coerce_to_expr(characters if characters is not None else " ")
return Expr(f.lpad(string.expr, count.expr, characters.expr))
def ltrim(arg: Expr) -> Expr:
"""Removes all characters, spaces by default, from the beginning of a string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [" a "]})
>>> trim_df = df.select(dfn.functions.ltrim(dfn.col("a")).alias("trimmed"))
>>> trim_df.collect_column("trimmed")[0].as_py()
'a '
"""
return Expr(f.ltrim(arg.expr))
def md5(arg: Expr) -> Expr:
"""Computes an MD5 128-bit checksum for a string expression.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.md5(dfn.col("a")).alias("md5"))
>>> result.collect_column("md5")[0].as_py()
'5d41402abc4b2a76b9719d911017c592'
"""
return Expr(f.md5(arg.expr))
def nanvl(x: Expr, y: Expr) -> Expr:
"""Returns ``x`` if ``x`` is not ``NaN``. Otherwise returns ``y``.
Args:
x: Expression to return when it is not NaN.
y: Fallback expression to return when ``x`` is NaN.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [np.nan, 1.0], "b": [0.0, 0.0]})
>>> nanvl_df = df.select(
... dfn.functions.nanvl(dfn.col("a"), dfn.col("b")).alias("nanvl"))
>>> nanvl_df.collect_column("nanvl")[0].as_py()
0.0
>>> nanvl_df.collect_column("nanvl")[1].as_py()
1.0
"""
return Expr(f.nanvl(x.expr, y.expr))
def nvl(x: Expr, y: Expr) -> Expr:
"""Returns ``x`` if ``x`` is not ``NULL``. Otherwise returns ``y``.
Args:
x: Expression to return when it is not NULL.
y: Fallback expression to return when ``x`` is NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [None, 1], "b": [0, 0]})
>>> nvl_df = df.select(
... dfn.functions.nvl(dfn.col("a"), dfn.col("b")).alias("nvl")
... )
>>> nvl_df.collect_column("nvl")[0].as_py()
0
>>> nvl_df.collect_column("nvl")[1].as_py()
1
"""
return Expr(f.nvl(x.expr, y.expr))
def nvl2(x: Expr, y: Expr, z: Expr) -> Expr:
"""Returns ``y`` if ``x`` is not NULL. Otherwise returns ``z``.
Args:
x: Expression to check for NULL.
y: Expression to return when ``x`` is not NULL.
z: Expression to return when ``x`` is NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [None, 1], "b": [10, 20], "c": [30, 40]})
>>> result = df.select(
... dfn.functions.nvl2(
... dfn.col("a"), dfn.col("b"), dfn.col("c")).alias("nvl2")
... )
>>> result.collect_column("nvl2")[0].as_py()
30
>>> result.collect_column("nvl2")[1].as_py()
20
"""
return Expr(f.nvl2(x.expr, y.expr, z.expr))
def octet_length(arg: Expr) -> Expr:
"""Returns the number of bytes of a string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.octet_length(dfn.col("a")).alias("len"))
>>> result.collect_column("len")[0].as_py()
5
"""
return Expr(f.octet_length(arg.expr))
def overlay(
string: Expr,
substring: Expr | str,
start: Expr | int,
length: Expr | int | None = None,
) -> Expr:
"""Replace a substring with a new substring.
Replace the substring of string that starts at the ``start``'th character and
extends for ``length`` characters with new substring.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["abcdef"]})
>>> result = df.select(
... dfn.functions.overlay(dfn.col("a"), "XY", 3, 2).alias("o"))
>>> result.collect_column("o")[0].as_py()
'abXYef'
"""
substring = coerce_to_expr(substring)
start = coerce_to_expr(start)
if length is None:
return Expr(f.overlay(string.expr, substring.expr, start.expr))
length = coerce_to_expr(length)
return Expr(f.overlay(string.expr, substring.expr, start.expr, length.expr))
def pi() -> Expr:
"""Returns an approximate value of π.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> from math import pi
>>> result = df.select(
... dfn.functions.pi().alias("pi")
... )
>>> result.collect_column("pi")[0].as_py() == pi
True
"""
return Expr(f.pi())
def position(string: Expr, substring: Expr | str) -> Expr:
"""Finds the position from where the ``substring`` matches the ``string``.
See Also:
This is an alias for :py:func:`strpos`.
"""
return strpos(string, substring)
def power(base: Expr, exponent: Expr | int | float) -> Expr: # noqa: PYI041
"""Returns ``base`` raised to the power of ``exponent``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [2.0]})
>>> result = df.select(
... dfn.functions.power(dfn.col("a"), 3.0).alias("pow")
... )
>>> result.collect_column("pow")[0].as_py()
8.0
"""
exponent = coerce_to_expr(exponent)
return Expr(f.power(base.expr, exponent.expr))
def pow(base: Expr, exponent: Expr | int | float) -> Expr: # noqa: PYI041
"""Returns ``base`` raised to the power of ``exponent``.
See Also:
This is an alias of :py:func:`power`.
"""
return power(base, exponent)
def radians(arg: Expr) -> Expr:
"""Converts the argument from degrees to radians.
Examples:
>>> from math import pi
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [180.0]})
>>> result = df.select(
... dfn.functions.radians(dfn.col("a")).alias("rad")
... )
>>> result.collect_column("rad")[0].as_py() == pi
True
"""
return Expr(f.radians(arg.expr))
def regexp_like(
string: Expr, regex: Expr | str, flags: Expr | str | None = None
) -> Expr:
r"""Find if any regular expression (regex) matches exist.
Tests a string using a regular expression returning true if at least one match,
false otherwise.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello123"]})
>>> result = df.select(
... dfn.functions.regexp_like(dfn.col("a"), "\\d+").alias("m")
... )
>>> result.collect_column("m")[0].as_py()
True
Use ``flags`` for case-insensitive matching:
>>> result = df.select(
... dfn.functions.regexp_like(
... dfn.col("a"), "HELLO", flags="i",
... ).alias("m")
... )
>>> result.collect_column("m")[0].as_py()
True
"""
regex = coerce_to_expr(regex)
flags = coerce_to_expr_or_none(flags)
return Expr(
f.regexp_like(
string.expr, regex.expr, flags.expr if flags is not None else None
)
)
def regexp_match(
string: Expr, regex: Expr | str, flags: Expr | str | None = None
) -> Expr:
r"""Perform regular expression (regex) matching.
Returns an array with each element containing the leftmost-first match of the
corresponding index in ``regex`` to string in ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello 42 world"]})
>>> result = df.select(
... dfn.functions.regexp_match(dfn.col("a"), "(\\d+)").alias("m")
... )
>>> result.collect_column("m")[0].as_py()
['42']
Use ``flags`` for case-insensitive matching:
>>> result = df.select(
... dfn.functions.regexp_match(
... dfn.col("a"), "(HELLO)", flags="i",
... ).alias("m")
... )
>>> result.collect_column("m")[0].as_py()
['hello']
"""
regex = coerce_to_expr(regex)
flags = coerce_to_expr_or_none(flags)
return Expr(
f.regexp_match(
string.expr, regex.expr, flags.expr if flags is not None else None
)
)
def regexp_replace(
string: Expr,
pattern: Expr | str,
replacement: Expr | str,
flags: Expr | str | None = None,
) -> Expr:
r"""Replaces substring(s) matching a PCRE-like regular expression.
The full list of supported features and syntax can be found at
<https://docs.rs/regex/latest/regex/#syntax>
Supported flags with the addition of 'g' can be found at
<https://docs.rs/regex/latest/regex/#grouping-and-flags>
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello 42"]})
>>> result = df.select(
... dfn.functions.regexp_replace(
... dfn.col("a"), "\\d+", "XX"
... ).alias("r")
... )
>>> result.collect_column("r")[0].as_py()
'hello XX'
Use the ``g`` flag to replace all occurrences:
>>> df = ctx.from_pydict({"a": ["a1 b2 c3"]})
>>> result = df.select(
... dfn.functions.regexp_replace(
... dfn.col("a"), "\\d+", "X", flags="g",
... ).alias("r")
... )
>>> result.collect_column("r")[0].as_py()
'aX bX cX'
"""
pattern = coerce_to_expr(pattern)
replacement = coerce_to_expr(replacement)
flags = coerce_to_expr_or_none(flags)
return Expr(
f.regexp_replace(
string.expr,
pattern.expr,
replacement.expr,
flags.expr if flags is not None else None,
)
)
def regexp_count(
string: Expr,
pattern: Expr | str,
start: Expr | int | None = None,
flags: Expr | str | None = None,
) -> Expr:
"""Returns the number of matches in a string.
Optional start position (the first position is 1) to search for the regular
expression.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["abcabc"]})
>>> result = df.select(
... dfn.functions.regexp_count(dfn.col("a"), "abc").alias("c"))
>>> result.collect_column("c")[0].as_py()
2
Use ``start`` to begin searching from a position, and
``flags`` for case-insensitive matching:
>>> result = df.select(
... dfn.functions.regexp_count(
... dfn.col("a"), "ABC", start=4, flags="i",
... ).alias("c"))
>>> result.collect_column("c")[0].as_py()
1
"""
pattern = coerce_to_expr(pattern)
start = coerce_to_expr_or_none(start)
flags = coerce_to_expr_or_none(flags)
return Expr(
f.regexp_count(
string.expr,
pattern.expr,
start.expr if start is not None else None,
flags.expr if flags is not None else None,
)
)
def regexp_instr(
values: Expr,
regex: Expr | str,
start: Expr | int | None = None,
n: Expr | int | None = None,
flags: Expr | str | None = None,
sub_expr: Expr | int | None = None,
) -> Expr:
r"""Returns the position of a regular expression match in a string.
Args:
values: Data to search for the regular expression match.
regex: Regular expression to search for.
start: Optional position to start the search (the first position is 1).
n: Optional occurrence of the match to find (the first occurrence is 1).
flags: Optional regular expression flags to control regex behavior.
sub_expr: Optionally capture group position instead of the entire match.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello 42 world"]})
>>> result = df.select(
... dfn.functions.regexp_instr(dfn.col("a"), "\\d+").alias("pos")
... )
>>> result.collect_column("pos")[0].as_py()
7
Use ``start`` to search from a position, ``n`` for the
nth occurrence, and ``flags`` for case-insensitive mode:
>>> df = ctx.from_pydict({"a": ["abc ABC abc"]})
>>> result = df.select(
... dfn.functions.regexp_instr(
... dfn.col("a"), "abc",
... start=2, n=1, flags="i",
... ).alias("pos")
... )
>>> result.collect_column("pos")[0].as_py()
5
Use ``sub_expr`` to get the position of a capture group:
>>> result = df.select(
... dfn.functions.regexp_instr(
... dfn.col("a"), "(abc)", sub_expr=1,
... ).alias("pos")
... )
>>> result.collect_column("pos")[0].as_py()
1
"""
regex = coerce_to_expr(regex)
start = coerce_to_expr_or_none(start)
n = coerce_to_expr_or_none(n)
flags = coerce_to_expr_or_none(flags)
sub_expr = coerce_to_expr_or_none(sub_expr)
return Expr(
f.regexp_instr(
values.expr,
regex.expr,
start.expr if start is not None else None,
n.expr if n is not None else None,
flags.expr if flags is not None else None,
sub_expr.expr if sub_expr is not None else None,
)
)
def repeat(string: Expr, n: Expr | int) -> Expr:
"""Repeats the ``string`` to ``n`` times.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["ha"]})
>>> result = df.select(
... dfn.functions.repeat(dfn.col("a"), 3).alias("r"))
>>> result.collect_column("r")[0].as_py()
'hahaha'
"""
n = coerce_to_expr(n)
return Expr(f.repeat(string.expr, n.expr))
def replace(string: Expr, from_val: Expr | str, to_val: Expr | str) -> Expr:
"""Replaces all occurrences of ``from_val`` with ``to_val`` in the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello world"]})
>>> result = df.select(
... dfn.functions.replace(dfn.col("a"), "world", "there").alias("r"))
>>> result.collect_column("r")[0].as_py()
'hello there'
"""
from_val = coerce_to_expr(from_val)
to_val = coerce_to_expr(to_val)
return Expr(f.replace(string.expr, from_val.expr, to_val.expr))
def reverse(arg: Expr) -> Expr:
"""Reverse the string argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.reverse(dfn.col("a")).alias("r"))
>>> result.collect_column("r")[0].as_py()
'olleh'
"""
return Expr(f.reverse(arg.expr))
def right(string: Expr, n: Expr | int) -> Expr:
"""Returns the last ``n`` characters in the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.right(dfn.col("a"), 3).alias("r"))
>>> result.collect_column("r")[0].as_py()
'llo'
"""
n = coerce_to_expr(n)
return Expr(f.right(string.expr, n.expr))
def round(value: Expr, decimal_places: Expr | int | None = None) -> Expr:
"""Round the argument to the nearest integer.
If the optional ``decimal_places`` is specified, round to the nearest number of
decimal places. You can specify a negative number of decimal places. For example
``round(lit(125.2345), -2)`` would yield a value of ``100.0``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.567]})
>>> result = df.select(dfn.functions.round(dfn.col("a"), 2).alias("r"))
>>> result.collect_column("r")[0].as_py()
1.57
"""
decimal_places = coerce_to_expr(decimal_places if decimal_places is not None else 0)
return Expr(f.round(value.expr, decimal_places.expr))
def rpad(string: Expr, count: Expr | int, characters: Expr | str | None = None) -> Expr:
"""Add right padding to a string.
Extends the string to length length by appending the characters fill (a space
by default). If the string is already longer than length then it is truncated.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hi"]})
>>> result = df.select(
... dfn.functions.rpad(dfn.col("a"), 5, "!").alias("r"))
>>> result.collect_column("r")[0].as_py()
'hi!!!'
"""
count = coerce_to_expr(count)
characters = coerce_to_expr(characters if characters is not None else " ")
return Expr(f.rpad(string.expr, count.expr, characters.expr))
def rtrim(arg: Expr) -> Expr:
"""Removes all characters, spaces by default, from the end of a string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [" a "]})
>>> trim_df = df.select(dfn.functions.rtrim(dfn.col("a")).alias("trimmed"))
>>> trim_df.collect_column("trimmed")[0].as_py()
' a'
"""
return Expr(f.rtrim(arg.expr))
def sha224(arg: Expr) -> Expr:
"""Computes the SHA-224 hash of a binary string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.sha224(dfn.col("a")).alias("h")
... )
>>> result.collect_column("h")[0].as_py().hex()
'ea09ae9cc6768c50fcee903ed054556e5bfc8347907f12598aa24193'
"""
return Expr(f.sha224(arg.expr))
def sha256(arg: Expr) -> Expr:
"""Computes the SHA-256 hash of a binary string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.sha256(dfn.col("a")).alias("h")
... )
>>> result.collect_column("h")[0].as_py().hex()
'2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824'
"""
return Expr(f.sha256(arg.expr))
def sha384(arg: Expr) -> Expr:
"""Computes the SHA-384 hash of a binary string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.sha384(dfn.col("a")).alias("h")
... )
>>> result.collect_column("h")[0].as_py().hex()
'59e1748777448c69de6b800d7a33bbfb9ff1b...
"""
return Expr(f.sha384(arg.expr))
def sha512(arg: Expr) -> Expr:
"""Computes the SHA-512 hash of a binary string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.sha512(dfn.col("a")).alias("h")
... )
>>> result.collect_column("h")[0].as_py().hex()
'9b71d224bd62f3785d96d46ad3ea3d73319bfb...
"""
return Expr(f.sha512(arg.expr))
def signum(arg: Expr) -> Expr:
"""Returns the sign of the argument (-1, 0, +1).
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [-5.0, 0.0, 5.0]})
>>> result = df.select(dfn.functions.signum(dfn.col("a")).alias("s"))
>>> result.collect_column("s").to_pylist()
[-1.0, 0.0, 1.0]
"""
return Expr(f.signum(arg.expr))
def sin(arg: Expr) -> Expr:
"""Returns the sine of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.sin(dfn.col("a")).alias("sin"))
>>> result.collect_column("sin")[0].as_py()
0.0
"""
return Expr(f.sin(arg.expr))
def sinh(arg: Expr) -> Expr:
"""Returns the hyperbolic sine of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.sinh(dfn.col("a")).alias("sinh"))
>>> result.collect_column("sinh")[0].as_py()
0.0
"""
return Expr(f.sinh(arg.expr))
def split_part(string: Expr, delimiter: Expr | str, index: Expr | int) -> Expr:
"""Split a string and return one part.
Splits a string based on a delimiter and picks out the desired field based
on the index.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["a,b,c"]})
>>> result = df.select(
... dfn.functions.split_part(dfn.col("a"), ",", 2).alias("s"))
>>> result.collect_column("s")[0].as_py()
'b'
"""
delimiter = coerce_to_expr(delimiter)
index = coerce_to_expr(index)
return Expr(f.split_part(string.expr, delimiter.expr, index.expr))
def sqrt(arg: Expr) -> Expr:
"""Returns the square root of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [9.0]})
>>> result = df.select(dfn.functions.sqrt(dfn.col("a")).alias("sqrt"))
>>> result.collect_column("sqrt")[0].as_py()
3.0
"""
return Expr(f.sqrt(arg.expr))
def starts_with(string: Expr, prefix: Expr | str) -> Expr:
"""Returns true if string starts with prefix.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello_from_datafusion"]})
>>> result = df.select(
... dfn.functions.starts_with(dfn.col("a"), "hello").alias("sw"))
>>> result.collect_column("sw")[0].as_py()
True
"""
prefix = coerce_to_expr(prefix)
return Expr(f.starts_with(string.expr, prefix.expr))
def strpos(string: Expr, substring: Expr | str) -> Expr:
"""Finds the position from where the ``substring`` matches the ``string``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.strpos(dfn.col("a"), "llo").alias("pos"))
>>> result.collect_column("pos")[0].as_py()
3
"""
substring = coerce_to_expr(substring)
return Expr(f.strpos(string.expr, substring.expr))
def substr(string: Expr, position: Expr | int) -> Expr:
"""Substring from the ``position`` to the end.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.substr(dfn.col("a"), 3).alias("s"))
>>> result.collect_column("s")[0].as_py()
'llo'
"""
position = coerce_to_expr(position)
return Expr(f.substr(string.expr, position.expr))
def substr_index(string: Expr, delimiter: Expr | str, count: Expr | int) -> Expr:
"""Returns an indexed substring.
The return will be the ``string`` from before ``count`` occurrences of
``delimiter``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["a.b.c"]})
>>> result = df.select(
... dfn.functions.substr_index(dfn.col("a"), ".", 2).alias("s"))
>>> result.collect_column("s")[0].as_py()
'a.b'
"""
delimiter = coerce_to_expr(delimiter)
count = coerce_to_expr(count)
return Expr(f.substr_index(string.expr, delimiter.expr, count.expr))
def substring(string: Expr, position: Expr | int, length: Expr | int) -> Expr:
"""Substring from the ``position`` with ``length`` characters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello world"]})
>>> result = df.select(
... dfn.functions.substring(dfn.col("a"), 1, 5).alias("s"))
>>> result.collect_column("s")[0].as_py()
'hello'
"""
position = coerce_to_expr(position)
length = coerce_to_expr(length)
return Expr(f.substring(string.expr, position.expr, length.expr))
def tan(arg: Expr) -> Expr:
"""Returns the tangent of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.tan(dfn.col("a")).alias("tan"))
>>> result.collect_column("tan")[0].as_py()
0.0
"""
return Expr(f.tan(arg.expr))
def tanh(arg: Expr) -> Expr:
"""Returns the hyperbolic tangent of the argument.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0]})
>>> result = df.select(dfn.functions.tanh(dfn.col("a")).alias("tanh"))
>>> result.collect_column("tanh")[0].as_py()
0.0
"""
return Expr(f.tanh(arg.expr))
def to_hex(arg: Expr) -> Expr:
"""Converts an integer to a hexadecimal string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [255]})
>>> result = df.select(dfn.functions.to_hex(dfn.col("a")).alias("hex"))
>>> result.collect_column("hex")[0].as_py()
'ff'
"""
return Expr(f.to_hex(arg.expr))
def now() -> Expr:
"""Returns the current timestamp in nanoseconds.
This will use the same value for all instances of now() in same statement.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.now().alias("now")
... )
Use .value instead of .as_py() because nanosecond timestamps
require pandas to convert to Python datetime objects.
>>> result.collect_column("now")[0].value > 0
True
"""
return Expr(f.now())
def current_timestamp() -> Expr:
"""Returns the current timestamp in nanoseconds.
See Also:
This is an alias for :py:func:`now`.
"""
return now()
def to_char(arg: Expr, formatter: Expr | str) -> Expr:
"""Returns a string representation of a date, time, timestamp or duration.
For usage of ``formatter`` see the rust chrono package ``strftime`` package.
[Documentation here.](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_char(
... dfn.functions.to_timestamp(dfn.col("a")),
... "%Y/%m/%d",
... ).alias("formatted")
... )
>>> result.collect_column("formatted")[0].as_py()
'2021/01/01'
"""
formatter = coerce_to_expr(formatter)
return Expr(f.to_char(arg.expr, formatter.expr))
def date_format(arg: Expr, formatter: Expr | str) -> Expr:
"""Returns a string representation of a date, time, timestamp or duration.
See Also:
This is an alias for :py:func:`to_char`.
"""
return to_char(arg, formatter)
def _unwrap_exprs(args: Iterable[Expr]) -> list:
return [arg.expr for arg in args]
def to_date(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a value to a date (YYYY-MM-DD).
Supports strings, numeric and timestamp types as input.
Integers and doubles are interpreted as days since the unix epoch.
Strings are parsed as YYYY-MM-DD (e.g. '2023-07-20')
if ``formatters`` are not provided.
For usage of ``formatters`` see the rust chrono package ``strftime`` package.
[Documentation here.](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-07-20"]})
>>> result = df.select(
... dfn.functions.to_date(dfn.col("a")).alias("dt"))
>>> str(result.collect_column("dt")[0].as_py())
'2021-07-20'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["20-07-2021"]})
>>> result = df.select(
... dfn.functions.to_date(dfn.col("a"), "%d-%m-%Y").alias("dt"))
>>> str(result.collect_column("dt")[0].as_py())
'2021-07-20'
"""
return Expr(f.to_date(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters))))
def to_local_time(*args: Expr) -> Expr:
"""Converts a timestamp with a timezone to a timestamp without a timezone.
This function handles daylight saving time changes.
"""
return Expr(f.to_local_time(*_unwrap_exprs(args)))
def to_time(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a value to a time. Supports strings and timestamps as input.
If ``formatters`` is not provided strings are parsed as HH:MM:SS, HH:MM or
HH:MM:SS.nnnnnnnnn;
For usage of ``formatters`` see the rust chrono package ``strftime`` package.
[Documentation here.](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["14:30:00"]})
>>> result = df.select(
... dfn.functions.to_time(dfn.col("a")).alias("t"))
>>> str(result.collect_column("t")[0].as_py())
'14:30:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["14h30m00s"]})
>>> result = df.select(
... dfn.functions.to_time(dfn.col("a"), "%Hh%Mm%Ss").alias("t"))
>>> str(result.collect_column("t")[0].as_py())
'14:30:00'
"""
return Expr(f.to_time(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters))))
def to_timestamp(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a string and optional formats to a ``Timestamp`` in nanoseconds.
For usage of ``formatters`` see the rust chrono package ``strftime`` package.
[Documentation here.](https://docs.rs/chrono/latest/chrono/format/strftime/index.html)
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/2021 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
"""
return Expr(
f.to_timestamp(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters)))
)
def to_timestamp_millis(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a string and optional formats to a ``Timestamp`` in milliseconds.
See :py:func:`to_timestamp` for a description on how to use formatters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_millis(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/2021 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_millis(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
"""
return Expr(
f.to_timestamp_millis(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters)))
)
def to_timestamp_micros(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a string and optional formats to a ``Timestamp`` in microseconds.
See :py:func:`to_timestamp` for a description on how to use formatters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_micros(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/2021 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_micros(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
"""
return Expr(
f.to_timestamp_micros(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters)))
)
def to_timestamp_nanos(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a string and optional formats to a ``Timestamp`` in nanoseconds.
See :py:func:`to_timestamp` for a description on how to use formatters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_nanos(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/2021 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_nanos(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
"""
return Expr(
f.to_timestamp_nanos(arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters)))
)
def to_timestamp_seconds(arg: Expr, *formatters: Expr | str) -> Expr:
"""Converts a string and optional formats to a ``Timestamp`` in seconds.
See :py:func:`to_timestamp` for a description on how to use formatters.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-01-01T00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_seconds(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/2021 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_timestamp_seconds(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'2021-01-01 00:00:00'
"""
return Expr(
f.to_timestamp_seconds(
arg.expr, *_unwrap_exprs(coerce_to_expr_list(formatters))
)
)
def to_unixtime(string: Expr, *format_arguments: Expr | str) -> Expr:
"""Converts a string and optional formats to a Unixtime.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["1970-01-01T00:00:00"]})
>>> result = df.select(dfn.functions.to_unixtime(dfn.col("a")).alias("u"))
>>> result.collect_column("u")[0].as_py()
0
Pass a format string as a bare ``str``:
>>> df = ctx.from_pydict({"a": ["01/01/1970 00:00:00"]})
>>> result = df.select(
... dfn.functions.to_unixtime(
... dfn.col("a"), "%d/%m/%Y %H:%M:%S"
... ).alias("u")
... )
>>> result.collect_column("u")[0].as_py()
0
"""
return Expr(
f.to_unixtime(
string.expr, *_unwrap_exprs(coerce_to_expr_list(format_arguments))
)
)
def current_date() -> Expr:
"""Returns current UTC date as a Date32 value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.current_date().alias("d")
... )
>>> result.collect_column("d")[0].as_py() is not None
True
"""
return Expr(f.current_date())
today = current_date
def current_time() -> Expr:
"""Returns current UTC time as a Time64 value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.current_time().alias("t")
... )
Use .value instead of .as_py() because nanosecond timestamps
require pandas to convert to Python datetime objects.
>>> result.collect_column("t")[0].value > 0
True
"""
return Expr(f.current_time())
def datepart(part: Expr | str, date: Expr) -> Expr:
"""Return a specified part of a date.
See Also:
This is an alias for :py:func:`date_part`.
"""
return _date_part(part, date, "datepart")
def date_part(part: Expr | str, date: Expr) -> Expr:
"""Extracts a subfield from the date.
Args:
part: The part of the date to extract. Must be one of ``"year"``,
``"month"``, ``"day"``, ``"hour"``, ``"minute"``, ``"second"``, etc.
date: The date expression to extract from.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-07-15T00:00:00"]})
>>> df = df.select(dfn.functions.to_timestamp(dfn.col("a")).alias("a"))
>>> result = df.select(
... dfn.functions.date_part("year", dfn.col("a")).alias("y"))
>>> result.collect_column("y")[0].as_py()
2021
"""
return _date_part(part, date, "date_part")
def _date_part(part: Expr | str, date: Expr, function_name: str) -> Expr:
_warn_if_expr_for_literal_arg(part, function_name, "part")
part = coerce_to_expr(part)
return Expr(f.date_part(part.expr, date.expr))
def extract(part: Expr | str, date: Expr) -> Expr:
"""Extracts a subfield from the date.
See Also:
This is an alias for :py:func:`date_part`.
"""
return _date_part(part, date, "extract")
def date_trunc(part: Expr | str, date: Expr) -> Expr:
"""Truncates the date to a specified level of precision.
Args:
part: The precision to truncate to. Must be one of ``"year"``,
``"month"``, ``"day"``, ``"hour"``, ``"minute"``, ``"second"``, etc.
date: The date expression to truncate.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["2021-07-15T12:34:56"]})
>>> df = df.select(dfn.functions.to_timestamp(dfn.col("a")).alias("a"))
>>> result = df.select(
... dfn.functions.date_trunc("month", dfn.col("a")).alias("t")
... )
>>> str(result.collect_column("t")[0].as_py())
'2021-07-01 00:00:00'
"""
return _date_trunc(part, date, "date_trunc")
def _date_trunc(part: Expr | str, date: Expr, function_name: str) -> Expr:
_warn_if_expr_for_literal_arg(part, function_name, "part")
part = coerce_to_expr(part)
return Expr(f.date_trunc(part.expr, date.expr))
def datetrunc(part: Expr | str, date: Expr) -> Expr:
"""Truncates the date to a specified level of precision.
See Also:
This is an alias for :py:func:`date_trunc`.
"""
return _date_trunc(part, date, "datetrunc")
def date_bin(stride: Expr | str, source: Expr | str, origin: Expr | str) -> Expr:
"""Coerces an arbitrary timestamp to the start of the nearest specified interval.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"timestamp": ['2021-07-15 12:34:56', '2021-01-01']})
>>> result = df.select(
... dfn.functions.date_bin(
... "15 minutes",
... dfn.col("timestamp"),
... "2001-01-01 00:00:00",
... ).alias("b")
... )
>>> str(result.collect_column("b")[0].as_py())
'2021-07-15 12:30:00'
>>> str(result.collect_column("b")[1].as_py())
'2021-01-01 00:00:00'
``source`` may also be a bare literal:
>>> result = df.select(
... dfn.functions.date_bin(
... "15 minutes", "2021-07-15 12:34:56", "2001-01-01 00:00:00"
... ).alias("b")
... )
>>> str(result.collect_column("b")[0].as_py())
'2021-07-15 12:30:00'
"""
# date_bin's planner coerces Utf8 (not Utf8View) literals to Interval/Timestamp,
# so wrap bare strs via string_literal to force Utf8.
stride = Expr.string_literal(stride) if isinstance(stride, str) else stride
source = Expr.string_literal(source) if isinstance(source, str) else source
origin = Expr.string_literal(origin) if isinstance(origin, str) else origin
return Expr(f.date_bin(stride.expr, source.expr, origin.expr))
def make_date(year: Expr | int, month: Expr | int, day: Expr | int) -> Expr:
"""Make a date from year, month and day component parts.
Examples:
>>> from datetime import date
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [2024], "m": [1], "d": [15]})
>>> result = df.select(
... dfn.functions.make_date(dfn.col("y"), dfn.col("m"),
... dfn.col("d")).alias("dt"))
>>> result.collect_column("dt")[0].as_py()
datetime.date(2024, 1, 15)
Pass bare ints for any component:
>>> df = ctx.from_pydict({"y": [2024]})
>>> result = df.select(
... dfn.functions.make_date(dfn.col("y"), 1, 15).alias("dt"))
>>> result.collect_column("dt")[0].as_py()
datetime.date(2024, 1, 15)
"""
year = coerce_to_expr(year)
month = coerce_to_expr(month)
day = coerce_to_expr(day)
return Expr(f.make_date(year.expr, month.expr, day.expr))
def make_time(hour: Expr | int, minute: Expr | int, second: Expr | int) -> Expr:
"""Make a time from hour, minute and second component parts.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"h": [12], "m": [30], "s": [0]})
>>> result = df.select(
... dfn.functions.make_time(dfn.col("h"), dfn.col("m"),
... dfn.col("s")).alias("t"))
>>> result.collect_column("t")[0].as_py()
datetime.time(12, 30)
Pass bare ints for any component:
>>> df = ctx.from_pydict({"h": [12]})
>>> result = df.select(
... dfn.functions.make_time(dfn.col("h"), 30, 0).alias("t"))
>>> result.collect_column("t")[0].as_py()
datetime.time(12, 30)
"""
hour = coerce_to_expr(hour)
minute = coerce_to_expr(minute)
second = coerce_to_expr(second)
return Expr(f.make_time(hour.expr, minute.expr, second.expr))
def translate(string: Expr, from_val: Expr | str, to_val: Expr | str) -> Expr:
"""Replaces the characters in ``from_val`` with the counterpart in ``to_val``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(
... dfn.functions.translate(dfn.col("a"), "helo", "HELO").alias("t"))
>>> result.collect_column("t")[0].as_py()
'HELLO'
"""
from_val = coerce_to_expr(from_val)
to_val = coerce_to_expr(to_val)
return Expr(f.translate(string.expr, from_val.expr, to_val.expr))
def trim(arg: Expr) -> Expr:
"""Removes all characters, spaces by default, from both sides of a string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [" hello "]})
>>> result = df.select(dfn.functions.trim(dfn.col("a")).alias("t"))
>>> result.collect_column("t")[0].as_py()
'hello'
"""
return Expr(f.trim(arg.expr))
def trunc(num: Expr, precision: Expr | int | None = None) -> Expr:
"""Truncate the number toward zero with optional precision.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.567]})
>>> result = df.select(
... dfn.functions.trunc(dfn.col("a")).alias("t"))
>>> result.collect_column("t")[0].as_py()
1.0
>>> result = df.select(
... dfn.functions.trunc(dfn.col("a"), precision=2).alias("t"))
>>> result.collect_column("t")[0].as_py()
1.56
"""
if precision is not None:
precision = coerce_to_expr(precision)
return Expr(f.trunc(num.expr, precision.expr))
return Expr(f.trunc(num.expr))
def upper(arg: Expr) -> Expr:
"""Converts a string to uppercase.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello"]})
>>> result = df.select(dfn.functions.upper(dfn.col("a")).alias("u"))
>>> result.collect_column("u")[0].as_py()
'HELLO'
"""
return Expr(f.upper(arg.expr))
def make_array(*args: Expr) -> Expr:
"""Returns an array using the specified input expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.make_array(
... dfn.lit(1), dfn.lit(2), dfn.lit(3)
... ).alias("arr"))
>>> result.collect_column("arr")[0].as_py()
[1, 2, 3]
"""
args = [arg.expr for arg in args]
return Expr(f.make_array(args))
def make_list(*args: Expr) -> Expr:
"""Returns an array using the specified input expressions.
See Also:
This is an alias for :py:func:`make_array`.
"""
return make_array(*args)
def array(*args: Expr) -> Expr:
"""Returns an array using the specified input expressions.
See Also:
This is an alias for :py:func:`make_array`.
"""
return make_array(*args)
def range(start: Expr, stop: Expr, step: Expr) -> Expr:
"""Create a list of values in the range between start and stop.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.range(dfn.lit(0), dfn.lit(5), dfn.lit(2)).alias("r"))
>>> result.collect_column("r")[0].as_py()
[0, 2, 4]
"""
return Expr(f.range(start.expr, stop.expr, step.expr))
def uuid() -> Expr:
"""Returns uuid v4 as a string value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.uuid().alias("u")
... )
>>> len(result.collect_column("u")[0].as_py()) == 36
True
"""
return Expr(f.uuid())
def struct(*args: Expr) -> Expr:
"""Returns a struct with the given arguments.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1], "b": [2]})
>>> result = df.select(
... dfn.functions.struct(
... dfn.col("a"), dfn.col("b")
... ).alias("s")
... )
Children in the new struct will always be `c0`, ..., `cN-1`
for `N` children.
>>> result.collect_column("s")[0].as_py() == {"c0": 1, "c1": 2}
True
"""
args = [arg.expr for arg in args]
return Expr(f.struct(*args))
def named_struct(name_pairs: list[tuple[str, Expr]]) -> Expr:
"""Returns a struct with the given names and arguments pairs.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.named_struct(
... [("x", dfn.lit(10)), ("y", dfn.lit(20))]
... ).alias("s")
... )
>>> result.collect_column("s")[0].as_py() == {"x": 10, "y": 20}
True
"""
name_pair_exprs = [
[Expr.literal(pa.scalar(pair[0], type=pa.string())), pair[1]]
for pair in name_pairs
]
# flatten
name_pairs = [x.expr for xs in name_pair_exprs for x in xs]
return Expr(f.named_struct(*name_pairs))
def from_unixtime(arg: Expr) -> Expr:
"""Converts an integer to RFC3339 timestamp format string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0]})
>>> result = df.select(
... dfn.functions.from_unixtime(
... dfn.col("a")
... ).alias("ts")
... )
>>> str(result.collect_column("ts")[0].as_py())
'1970-01-01 00:00:00'
"""
return Expr(f.from_unixtime(arg.expr))
def arrow_typeof(arg: Expr) -> Expr:
"""Returns the Arrow type of the expression.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(dfn.functions.arrow_typeof(dfn.col("a")).alias("t"))
>>> result.collect_column("t")[0].as_py()
'Int64'
"""
return Expr(f.arrow_typeof(arg.expr))
def arrow_cast(expr: Expr, data_type: Expr | str | pa.DataType) -> Expr:
"""Casts an expression to a specified data type.
The ``data_type`` can be a string, a ``pyarrow.DataType``, or an
``Expr``. For simple types, :py:meth:`Expr.cast()
<datafusion.expr.Expr.cast>` is more concise
(e.g., ``col("a").cast(pa.float64())``). Use ``arrow_cast`` when
you want to specify the target type as a string using DataFusion's
type syntax, which can be more readable for complex types like
``"Timestamp(Nanosecond, None)"``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.arrow_cast(dfn.col("a"), "Float64").alias("c")
... )
>>> result.collect_column("c")[0].as_py()
1.0
>>> result = df.select(
... dfn.functions.arrow_cast(
... dfn.col("a"), data_type=pa.float64()
... ).alias("c")
... )
>>> result.collect_column("c")[0].as_py()
1.0
"""
_warn_if_expr_for_literal_arg(data_type, "arrow_cast", "data_type")
if isinstance(data_type, pa.DataType):
return expr.cast(data_type)
if isinstance(data_type, str):
data_type = Expr.string_literal(data_type)
return Expr(f.arrow_cast(expr.expr, data_type.expr))
def arrow_try_cast(expr: Expr, data_type: Expr | str | pa.DataType) -> Expr:
"""Casts an expression to a specified data type, returning NULL on failure.
Like :py:func:`arrow_cast` but produces NULL instead of erroring when the
cast cannot be performed. The ``data_type`` may be a string in DataFusion
type syntax (for example ``"Float64"``), a ``pyarrow.DataType``, or an
``Expr`` of string type.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["oops"]})
>>> result = df.select(
... dfn.functions.arrow_try_cast(dfn.col("a"), "Float64").alias("c")
... )
>>> result.collect_column("c")[0].as_py() is None
True
>>> result = df.select(
... dfn.functions.arrow_try_cast(
... dfn.col("a"), data_type=pa.float64()
... ).alias("c")
... )
>>> result.collect_column("c")[0].as_py() is None
True
"""
_warn_if_expr_for_literal_arg(data_type, "arrow_try_cast", "data_type")
if isinstance(data_type, pa.DataType):
return expr.try_cast(data_type)
if isinstance(data_type, str):
data_type = Expr.string_literal(data_type)
return Expr(f.arrow_try_cast(expr.expr, data_type.expr))
def arrow_field(expr: Expr) -> Expr:
"""Returns the Arrow field information of an expression as a struct.
The returned struct contains the field's name, data type, nullability,
and metadata.
Examples:
>>> field = pa.field("val", pa.int64(), metadata={"k": "v"})
>>> schema = pa.schema([field])
>>> batch = pa.RecordBatch.from_arrays([pa.array([1])], schema=schema)
>>> ctx = dfn.SessionContext()
>>> df = ctx.create_dataframe([[batch]])
>>> result = df.select(
... dfn.functions.arrow_field(dfn.col("val")).alias("f")
... )
>>> out = result.collect_column("f")[0].as_py()
>>> out["name"], out["data_type"], out["nullable"], out["metadata"]
('val', 'Int64', True, [('k', 'v')])
"""
return Expr(f.arrow_field(expr.expr))
def cast_to_type(value: Expr, type_ref: Expr) -> Expr:
"""Casts ``value`` to the data type of ``type_ref``.
Only the *type* of ``type_ref`` is used; its value is ignored. This is
useful when the target type comes from another column or expression
rather than being known up-front. Casts that fail produce an error; use
:py:func:`try_cast_to_type` for the NULL-on-failure variant.
If the target type is known statically, prefer :py:func:`arrow_cast`
(or :py:func:`arrow_try_cast` for the NULL-on-failure variant) and
pass a type string or ``pyarrow.DataType`` directly.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1], "b": [1.0]})
>>> result = df.select(
... dfn.functions.cast_to_type(
... dfn.col("a"), dfn.col("b")
... ).alias("c")
... )
>>> result.collect_column("c")[0].as_py()
1.0
"""
return Expr(f.cast_to_type(value.expr, type_ref.expr))
def try_cast_to_type(value: Expr, type_ref: Expr) -> Expr:
"""Casts ``value`` to the data type of ``type_ref``, NULL on failure.
Like :py:func:`cast_to_type`, but casts that fail produce NULL instead
of erroring. Only the *type* of ``type_ref`` is used; its value is
ignored.
If the target type is known statically, prefer :py:func:`arrow_try_cast`
and pass a type string or ``pyarrow.DataType`` directly.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["oops"], "b": [1.0]})
>>> result = df.select(
... dfn.functions.try_cast_to_type(
... dfn.col("a"), dfn.col("b")
... ).alias("c")
... )
>>> result.collect_column("c")[0].as_py() is None
True
"""
return Expr(f.try_cast_to_type(value.expr, type_ref.expr))
def arrow_metadata(expr: Expr, key: Expr | str | None = None) -> Expr:
"""Returns the metadata of the input expression.
If called with one argument, returns a Map of all metadata key-value pairs.
If called with two arguments, returns the value for the specified metadata key.
Examples:
>>> field = pa.field("val", pa.int64(), metadata={"k": "v"})
>>> schema = pa.schema([field])
>>> batch = pa.RecordBatch.from_arrays([pa.array([1])], schema=schema)
>>> ctx = dfn.SessionContext()
>>> df = ctx.create_dataframe([[batch]])
>>> result = df.select(
... dfn.functions.arrow_metadata(dfn.col("val")).alias("meta")
... )
>>> ("k", "v") in result.collect_column("meta")[0].as_py()
True
>>> result = df.select(
... dfn.functions.arrow_metadata(
... dfn.col("val"), key="k"
... ).alias("meta_val")
... )
>>> result.collect_column("meta_val")[0].as_py()
'v'
"""
if key is None:
return Expr(f.arrow_metadata(expr.expr))
_warn_if_expr_for_literal_arg(key, "arrow_metadata", "key")
if isinstance(key, str):
key = Expr.string_literal(key)
return Expr(f.arrow_metadata(expr.expr, key.expr))
def with_metadata(expr: Expr, metadata: dict[str, str]) -> Expr:
"""Attaches Arrow field metadata (key/value pairs) to the input expression.
This is the inverse of :py:func:`arrow_metadata`. Existing metadata on the
input field is preserved; new keys overwrite on collision. Keys must be
non-empty strings; empty values are allowed.
An empty ``metadata`` dict is a no-op and returns the input expression
unchanged. Empty keys raise :py:class:`ValueError`.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.with_metadata(
... dfn.col("a"), {"unit": "ms"}
... ).alias("a")
... )
>>> result.select(
... dfn.functions.arrow_metadata(dfn.col("a"), "unit").alias("u")
... ).collect_column("u")[0].as_py()
'ms'
"""
if not metadata:
return expr
args = [expr.expr]
for k, v in metadata.items():
if not k:
msg = "with_metadata keys must be non-empty strings"
raise ValueError(msg)
args.append(Expr.string_literal(k).expr)
args.append(Expr.string_literal(v).expr)
return Expr(f.with_metadata(*args))
def get_field(expr: Expr, *names: Expr | str) -> Expr:
"""Extracts a (possibly nested) field from a struct or map by name.
Pass one name for a single-level lookup, or several names to walk a path
of nested struct/map fields in a single ``get_field`` call. For a single
static-string name, ``expr["field"]`` is a convenient shorthand; use
``get_field`` when the field name is a dynamic
:py:class:`~datafusion.expr.Expr` or when traversing multiple levels at
once.
Args:
expr: The struct or map expression to read from.
*names: One or more field names (``str``) or expressions
(:py:class:`~datafusion.expr.Expr`).
Examples:
Single-level lookup:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1], "b": [2]})
>>> df = df.with_column(
... "s",
... F.named_struct([("x", dfn.col("a")), ("y", dfn.col("b"))]),
... )
>>> result = df.select(
... F.get_field(dfn.col("s"), "x").alias("x_val")
... )
>>> result.collect_column("x_val")[0].as_py()
1
Equivalent using bracket syntax:
>>> result = df.select(
... dfn.col("s")["x"].alias("x_val")
... )
>>> result.collect_column("x_val")[0].as_py()
1
Multi-level lookup:
>>> df = df.with_column(
... "outer",
... F.named_struct([("inner", dfn.col("s"))]),
... )
>>> result = df.select(
... F.get_field(dfn.col("outer"), "inner", "x").alias("x_val")
... )
>>> result.collect_column("x_val")[0].as_py()
1
"""
if not names:
msg = "get_field requires at least one field name"
raise ValueError(msg)
resolved = [Expr.string_literal(n) if isinstance(n, str) else n for n in names]
return Expr(f.get_field(expr.expr, [n.expr for n in resolved]))
def union_extract(union_expr: Expr, field_name: Expr | str) -> Expr:
"""Extracts a value from a union type by field name.
Returns the value of the named field if it is the currently selected
variant, otherwise returns NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> types = pa.array([0, 1, 0], type=pa.int8())
>>> offsets = pa.array([0, 0, 1], type=pa.int32())
>>> arr = pa.UnionArray.from_dense(
... types, offsets, [pa.array([1, 2]), pa.array(["hi"])],
... ["int", "str"], [0, 1],
... )
>>> batch = pa.RecordBatch.from_arrays([arr], names=["u"])
>>> df = ctx.create_dataframe([[batch]])
>>> result = df.select(
... dfn.functions.union_extract(dfn.col("u"), "int").alias("val")
... )
>>> result.collect_column("val").to_pylist()
[1, None, 2]
"""
if isinstance(field_name, str):
field_name = Expr.string_literal(field_name)
return Expr(f.union_extract(union_expr.expr, field_name.expr))
def union_tag(union_expr: Expr) -> Expr:
"""Returns the tag (active field name) of a union type.
Examples:
>>> ctx = dfn.SessionContext()
>>> types = pa.array([0, 1, 0], type=pa.int8())
>>> offsets = pa.array([0, 0, 1], type=pa.int32())
>>> arr = pa.UnionArray.from_dense(
... types, offsets, [pa.array([1, 2]), pa.array(["hi"])],
... ["int", "str"], [0, 1],
... )
>>> batch = pa.RecordBatch.from_arrays([arr], names=["u"])
>>> df = ctx.create_dataframe([[batch]])
>>> result = df.select(
... dfn.functions.union_tag(dfn.col("u")).alias("tag")
... )
>>> result.collect_column("tag").to_pylist()
['int', 'str', 'int']
"""
return Expr(f.union_tag(union_expr.expr))
def version() -> Expr:
"""Returns the DataFusion version string.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.empty_table()
>>> result = df.select(dfn.functions.version().alias("v"))
>>> "Apache DataFusion" in result.collect_column("v")[0].as_py()
True
"""
return Expr(f.version())
def row(*args: Expr) -> Expr:
"""Returns a struct with the given arguments.
See Also:
This is an alias for :py:func:`struct`.
"""
return struct(*args)
def random() -> Expr:
"""Returns a random value in the range ``0.0 <= x < 1.0``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.random().alias("r")
... )
>>> val = result.collect_column("r")[0].as_py()
>>> 0.0 <= val < 1.0
True
"""
return Expr(f.random())
def array_append(array: Expr, element: Expr) -> Expr:
"""Appends an element to the end of an array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_append(dfn.col("a"), dfn.lit(4)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 3, 4]
"""
return Expr(f.array_append(array.expr, element.expr))
def array_push_back(array: Expr, element: Expr) -> Expr:
"""Appends an element to the end of an array.
See Also:
This is an alias for :py:func:`array_append`.
"""
return array_append(array, element)
def list_append(array: Expr, element: Expr) -> Expr:
"""Appends an element to the end of an array.
See Also:
This is an alias for :py:func:`array_append`.
"""
return array_append(array, element)
def list_push_back(array: Expr, element: Expr) -> Expr:
"""Appends an element to the end of an array.
See Also:
This is an alias for :py:func:`array_append`.
"""
return array_append(array, element)
def array_concat(*args: Expr) -> Expr:
"""Concatenates the input arrays.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2]], "b": [[3, 4]]})
>>> result = df.select(
... dfn.functions.array_concat(dfn.col("a"), dfn.col("b")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 3, 4]
"""
args = [arg.expr for arg in args]
return Expr(f.array_concat(args))
def array_cat(*args: Expr) -> Expr:
"""Concatenates the input arrays.
See Also:
This is an alias for :py:func:`array_concat`.
"""
return array_concat(*args)
def array_dims(array: Expr) -> Expr:
"""Returns an array of the array's dimensions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(dfn.functions.array_dims(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[3]
"""
return Expr(f.array_dims(array.expr))
def array_distinct(array: Expr) -> Expr:
"""Returns distinct values from the array after removing duplicates.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_distinct(
... dfn.col("a")
... ).alias("result")
... )
>>> sorted(
... result.collect_column("result")[0].as_py()
... )
[1, 2, 3]
"""
return Expr(f.array_distinct(array.expr))
def array_compact(array: Expr) -> Expr:
"""Removes NULL values from the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, None, 2, None, 3]]})
>>> result = df.select(
... dfn.functions.array_compact(dfn.col("a")).alias("result")
... )
>>> result.collect_column("result")[0].as_py()
[1, 2, 3]
"""
return Expr(f.array_compact(array.expr))
def array_normalize(array: Expr) -> Expr:
"""Scales a numeric array so it has Euclidean length 1.
Treats the array as a vector and divides every element by the vector's
Euclidean (L2) norm — the square root of the sum of the squared
elements. The returned array points in the same direction as the input
but has a magnitude of 1, which makes it suitable for cosine-similarity
comparisons and other operations that expect unit vectors.
For the input ``[3.0, 4.0]`` the L2 norm is ``sqrt(3**2 + 4**2) = 5``,
so each element is divided by 5 to produce ``[0.6, 0.8]``.
Normalizing the zero vector is undefined (it would divide by zero), so
the function returns NULL for an all-zero input. NULL is also returned
if any element of the input array is NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[3.0, 4.0]]})
>>> result = df.select(
... dfn.functions.array_normalize(dfn.col("a")).alias("result")
... )
>>> result.collect_column("result")[0].as_py()
[0.6, 0.8]
The zero vector has no direction to preserve, so the result is NULL:
>>> df_zero = ctx.from_pydict({"a": [[0.0, 0.0]]})
>>> result = df_zero.select(
... dfn.functions.array_normalize(dfn.col("a")).alias("result")
... )
>>> result.collect_column("result")[0].as_py() is None
True
"""
return Expr(f.array_normalize(array.expr))
def cosine_distance(array1: Expr, array2: Expr) -> Expr:
"""Measures how much two numeric arrays differ in direction.
Treats each input as a vector and compares the angle between them,
ignoring their magnitudes. The result is ``1 - cosine_similarity``,
where cosine similarity is the dot product of the two vectors divided
by the product of their Euclidean (L2) norms.
The returned value ranges from 0 to 2:
* ``0`` — vectors point in the same direction (any positive scaling
of one yields the other).
* ``1`` — vectors are orthogonal (no shared direction).
* ``2`` — vectors point in exactly opposite directions.
This is the standard distance metric for comparing embedding vectors
(text, image, audio) where direction carries the meaning and overall
magnitude does not.
Both arrays must have the same length; otherwise execution fails. If
either input is the zero vector the cosine is undefined and the
function returns NULL.
Examples:
Identical vectors have distance ``0``:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict(
... {"a": [[1.0, 2.0, 3.0]], "b": [[1.0, 2.0, 3.0]]}
... )
>>> result = df.select(
... dfn.functions.cosine_distance(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> result.collect_column("result")[0].as_py()
0.0
Orthogonal vectors have distance ``1``:
>>> df_orth = ctx.from_pydict(
... {"a": [[1.0, 0.0]], "b": [[0.0, 1.0]]}
... )
>>> result = df_orth.select(
... dfn.functions.cosine_distance(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> result.collect_column("result")[0].as_py()
1.0
"""
return Expr(f.cosine_distance(array1.expr, array2.expr))
def inner_product(array1: Expr, array2: Expr) -> Expr:
"""Returns the inner (dot) product of two numeric arrays.
Treats each input as a vector and returns the sum of the element-wise
products: ``sum(array1[i] * array2[i])``. For ``[1, 2, 3]`` and
``[4, 5, 6]`` the result is ``1*4 + 2*5 + 3*6 = 32``.
Also available as :py:func:`dot_product` (and as ``dot_product`` in
raw SQL).
Both arrays must have the same length; otherwise execution fails. NULL
is returned when either input array is NULL or when any element of
either array is NULL.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict(
... {"a": [[1.0, 2.0, 3.0]], "b": [[4.0, 5.0, 6.0]]}
... )
>>> result = df.select(
... dfn.functions.inner_product(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> result.collect_column("result")[0].as_py()
32.0
NULL elements propagate to NULL output:
>>> df_null = ctx.from_pydict(
... {"a": [[1.0, None, 3.0]], "b": [[4.0, 5.0, 6.0]]}
... )
>>> result = df_null.select(
... dfn.functions.inner_product(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> result.collect_column("result")[0].as_py() is None
True
"""
return Expr(f.inner_product(array1.expr, array2.expr))
def dot_product(array1: Expr, array2: Expr) -> Expr:
"""Returns the inner (dot) product of two numeric arrays.
See Also:
This is an alias for :py:func:`inner_product`.
"""
return inner_product(array1, array2)
def list_cat(*args: Expr) -> Expr:
"""Concatenates the input arrays.
See Also:
This is an alias for :py:func:`array_concat`, :py:func:`array_cat`.
"""
return array_concat(*args)
def list_concat(*args: Expr) -> Expr:
"""Concatenates the input arrays.
See Also:
This is an alias for :py:func:`array_concat`, :py:func:`array_cat`.
"""
return array_concat(*args)
def list_distinct(array: Expr) -> Expr:
"""Returns distinct values from the array after removing duplicates.
See Also:
This is an alias for :py:func:`array_distinct`.
"""
return array_distinct(array)
def list_compact(array: Expr) -> Expr:
"""Removes NULL values from the array.
See Also:
This is an alias for :py:func:`array_compact`.
"""
return array_compact(array)
def list_normalize(array: Expr) -> Expr:
"""Scales a numeric array so it has Euclidean length 1.
See Also:
This is an alias for :py:func:`array_normalize`.
"""
return array_normalize(array)
def list_dims(array: Expr) -> Expr:
"""Returns an array of the array's dimensions.
See Also:
This is an alias for :py:func:`array_dims`.
"""
return array_dims(array)
def array_element(array: Expr, n: Expr | int) -> Expr:
"""Extracts the element with the index n from the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[10, 20, 30]]})
>>> result = df.select(
... dfn.functions.array_element(dfn.col("a"), 2).alias("result"))
>>> result.collect_column("result")[0].as_py()
20
"""
n = coerce_to_expr(n)
return Expr(f.array_element(array.expr, n.expr))
def array_empty(array: Expr) -> Expr:
"""Returns a boolean indicating whether the array is empty.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2]]})
>>> result = df.select(dfn.functions.array_empty(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
False
"""
return Expr(f.array_empty(array.expr))
def list_empty(array: Expr) -> Expr:
"""Returns a boolean indicating whether the array is empty.
See Also:
This is an alias for :py:func:`array_empty`.
"""
return array_empty(array)
def array_extract(array: Expr, n: Expr | int) -> Expr:
"""Extracts the element with the index n from the array.
See Also:
This is an alias for :py:func:`array_element`.
"""
return array_element(array, n)
def list_element(array: Expr, n: Expr | int) -> Expr:
"""Extracts the element with the index n from the array.
See Also:
This is an alias for :py:func:`array_element`.
"""
return array_element(array, n)
def list_extract(array: Expr, n: Expr | int) -> Expr:
"""Extracts the element with the index n from the array.
See Also:
This is an alias for :py:func:`array_element`.
"""
return array_element(array, n)
def array_length(array: Expr) -> Expr:
"""Returns the length of the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(dfn.functions.array_length(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
3
"""
return Expr(f.array_length(array.expr))
def list_length(array: Expr) -> Expr:
"""Returns the length of the array.
See Also:
This is an alias for :py:func:`array_length`.
"""
return array_length(array)
def array_has(first_array: Expr, second_array: Expr) -> Expr:
"""Returns true if the element appears in the first array, otherwise false.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_has(dfn.col("a"), dfn.lit(2)).alias("result"))
>>> result.collect_column("result")[0].as_py()
True
"""
return Expr(f.array_has(first_array.expr, second_array.expr))
def array_has_all(first_array: Expr, second_array: Expr) -> Expr:
"""Determines if there is complete overlap ``second_array`` in ``first_array``.
Returns true if each element of the second array appears in the first array.
Otherwise, it returns false.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]], "b": [[1, 2]]})
>>> result = df.select(
... dfn.functions.array_has_all(dfn.col("a"), dfn.col("b")).alias("result"))
>>> result.collect_column("result")[0].as_py()
True
"""
return Expr(f.array_has_all(first_array.expr, second_array.expr))
def array_has_any(first_array: Expr, second_array: Expr) -> Expr:
"""Determine if there is an overlap between ``first_array`` and ``second_array``.
Returns true if at least one element of the second array appears in the first
array. Otherwise, it returns false.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]], "b": [[2, 5]]})
>>> result = df.select(
... dfn.functions.array_has_any(dfn.col("a"), dfn.col("b")).alias("result"))
>>> result.collect_column("result")[0].as_py()
True
"""
return Expr(f.array_has_any(first_array.expr, second_array.expr))
def array_contains(array: Expr, element: Expr) -> Expr:
"""Returns true if the element appears in the array, otherwise false.
See Also:
This is an alias for :py:func:`array_has`.
"""
return array_has(array, element)
def list_has(array: Expr, element: Expr) -> Expr:
"""Returns true if the element appears in the array, otherwise false.
See Also:
This is an alias for :py:func:`array_has`.
"""
return array_has(array, element)
def list_has_all(first_array: Expr, second_array: Expr) -> Expr:
"""Determines if there is complete overlap ``second_array`` in ``first_array``.
See Also:
This is an alias for :py:func:`array_has_all`.
"""
return array_has_all(first_array, second_array)
def list_has_any(first_array: Expr, second_array: Expr) -> Expr:
"""Determine if there is an overlap between ``first_array`` and ``second_array``.
See Also:
This is an alias for :py:func:`array_has_any`.
"""
return array_has_any(first_array, second_array)
def arrays_overlap(first_array: Expr, second_array: Expr) -> Expr:
"""Returns true if any element appears in both arrays.
See Also:
This is an alias for :py:func:`array_has_any`.
"""
return array_has_any(first_array, second_array)
def list_overlap(first_array: Expr, second_array: Expr) -> Expr:
"""Returns true if any element appears in both arrays.
See Also:
This is an alias for :py:func:`array_has_any`.
"""
return array_has_any(first_array, second_array)
def list_contains(array: Expr, element: Expr) -> Expr:
"""Returns true if the element appears in the array, otherwise false.
See Also:
This is an alias for :py:func:`array_has`.
"""
return array_has(array, element)
def array_position(array: Expr, element: Expr, index: int | None = 1) -> Expr:
"""Return the position of the first occurrence of ``element`` in ``array``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[10, 20, 30]]})
>>> result = df.select(
... dfn.functions.array_position(
... dfn.col("a"), dfn.lit(20)
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
2
Use ``index`` to start searching from a given position:
>>> df = ctx.from_pydict({"a": [[10, 20, 10, 20]]})
>>> result = df.select(
... dfn.functions.array_position(
... dfn.col("a"), dfn.lit(20), index=3,
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
4
"""
return Expr(f.array_position(array.expr, element.expr, index))
def array_indexof(array: Expr, element: Expr, index: int | None = 1) -> Expr:
"""Return the position of the first occurrence of ``element`` in ``array``.
See Also:
This is an alias for :py:func:`array_position`.
"""
return array_position(array, element, index)
def list_position(array: Expr, element: Expr, index: int | None = 1) -> Expr:
"""Return the position of the first occurrence of ``element`` in ``array``.
See Also:
This is an alias for :py:func:`array_position`.
"""
return array_position(array, element, index)
def list_indexof(array: Expr, element: Expr, index: int | None = 1) -> Expr:
"""Return the position of the first occurrence of ``element`` in ``array``.
See Also:
This is an alias for :py:func:`array_position`.
"""
return array_position(array, element, index)
def array_positions(array: Expr, element: Expr) -> Expr:
"""Searches for an element in the array and returns all occurrences.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1]]})
>>> result = df.select(
... dfn.functions.array_positions(dfn.col("a"), dfn.lit(1)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 3]
"""
return Expr(f.array_positions(array.expr, element.expr))
def list_positions(array: Expr, element: Expr) -> Expr:
"""Searches for an element in the array and returns all occurrences.
See Also:
This is an alias for :py:func:`array_positions`.
"""
return array_positions(array, element)
def array_ndims(array: Expr) -> Expr:
"""Returns the number of dimensions of the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(dfn.functions.array_ndims(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
1
"""
return Expr(f.array_ndims(array.expr))
def list_ndims(array: Expr) -> Expr:
"""Returns the number of dimensions of the array.
See Also:
This is an alias for :py:func:`array_ndims`.
"""
return array_ndims(array)
def array_prepend(element: Expr, array: Expr) -> Expr:
"""Prepends an element to the beginning of an array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2]]})
>>> result = df.select(
... dfn.functions.array_prepend(dfn.lit(0), dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[0, 1, 2]
"""
return Expr(f.array_prepend(element.expr, array.expr))
def array_push_front(element: Expr, array: Expr) -> Expr:
"""Prepends an element to the beginning of an array.
See Also:
This is an alias for :py:func:`array_prepend`.
"""
return array_prepend(element, array)
def list_prepend(element: Expr, array: Expr) -> Expr:
"""Prepends an element to the beginning of an array.
See Also:
This is an alias for :py:func:`array_prepend`.
"""
return array_prepend(element, array)
def list_push_front(element: Expr, array: Expr) -> Expr:
"""Prepends an element to the beginning of an array.
See Also:
This is an alias for :py:func:`array_prepend`.
"""
return array_prepend(element, array)
def array_pop_back(array: Expr) -> Expr:
"""Returns the array without the last element.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_pop_back(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2]
"""
return Expr(f.array_pop_back(array.expr))
def array_pop_front(array: Expr) -> Expr:
"""Returns the array without the first element.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_pop_front(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[2, 3]
"""
return Expr(f.array_pop_front(array.expr))
def list_pop_back(array: Expr) -> Expr:
"""Returns the array without the last element.
See Also:
This is an alias for :py:func:`array_pop_back`.
"""
return array_pop_back(array)
def list_pop_front(array: Expr) -> Expr:
"""Returns the array without the first element.
See Also:
This is an alias for :py:func:`array_pop_front`.
"""
return array_pop_front(array)
def array_remove(array: Expr, element: Expr) -> Expr:
"""Removes the first element from the array equal to the given value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1]]})
>>> result = df.select(
... dfn.functions.array_remove(dfn.col("a"), dfn.lit(1)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[2, 1]
"""
return Expr(f.array_remove(array.expr, element.expr))
def list_remove(array: Expr, element: Expr) -> Expr:
"""Removes the first element from the array equal to the given value.
See Also:
This is an alias for :py:func:`array_remove`.
"""
return array_remove(array, element)
def array_remove_n(array: Expr, element: Expr, max: Expr | int) -> Expr:
"""Removes the first ``max`` elements from the array equal to the given value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1, 1]]})
>>> result = df.select(
... dfn.functions.array_remove_n(
... dfn.col("a"), dfn.lit(1), 2
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[2, 1]
"""
max = coerce_to_expr(max)
return Expr(f.array_remove_n(array.expr, element.expr, max.expr))
def list_remove_n(array: Expr, element: Expr, max: Expr | int) -> Expr:
"""Removes the first ``max`` elements from the array equal to the given value.
See Also:
This is an alias for :py:func:`array_remove_n`.
"""
return array_remove_n(array, element, max)
def array_remove_all(array: Expr, element: Expr) -> Expr:
"""Removes all elements from the array equal to the given value.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1]]})
>>> result = df.select(
... dfn.functions.array_remove_all(
... dfn.col("a"), dfn.lit(1)
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[2]
"""
return Expr(f.array_remove_all(array.expr, element.expr))
def list_remove_all(array: Expr, element: Expr) -> Expr:
"""Removes all elements from the array equal to the given value.
See Also:
This is an alias for :py:func:`array_remove_all`.
"""
return array_remove_all(array, element)
def array_repeat(element: Expr, count: Expr | int) -> Expr:
"""Returns an array containing ``element`` ``count`` times.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.array_repeat(dfn.lit(3), 3).alias("result"))
>>> result.collect_column("result")[0].as_py()
[3, 3, 3]
"""
count = coerce_to_expr(count)
return Expr(f.array_repeat(element.expr, count.expr))
def list_repeat(element: Expr, count: Expr | int) -> Expr:
"""Returns an array containing ``element`` ``count`` times.
See Also:
This is an alias for :py:func:`array_repeat`.
"""
return array_repeat(element, count)
def array_replace(array: Expr, from_val: Expr, to_val: Expr) -> Expr:
"""Replaces the first occurrence of ``from_val`` with ``to_val``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1]]})
>>> result = df.select(
... dfn.functions.array_replace(dfn.col("a"), dfn.lit(1),
... dfn.lit(9)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[9, 2, 1]
"""
return Expr(f.array_replace(array.expr, from_val.expr, to_val.expr))
def list_replace(array: Expr, from_val: Expr, to_val: Expr) -> Expr:
"""Replaces the first occurrence of ``from_val`` with ``to_val``.
See Also:
This is an alias for :py:func:`array_replace`.
"""
return array_replace(array, from_val, to_val)
def array_replace_n(array: Expr, from_val: Expr, to_val: Expr, max: Expr | int) -> Expr:
"""Replace ``n`` occurrences of ``from_val`` with ``to_val``.
Replaces the first ``max`` occurrences of the specified element with another
specified element.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1, 1]]})
>>> result = df.select(
... dfn.functions.array_replace_n(
... dfn.col("a"), dfn.lit(1), dfn.lit(9), 2
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[9, 2, 9, 1]
"""
max = coerce_to_expr(max)
return Expr(f.array_replace_n(array.expr, from_val.expr, to_val.expr, max.expr))
def list_replace_n(array: Expr, from_val: Expr, to_val: Expr, max: Expr | int) -> Expr:
"""Replace ``n`` occurrences of ``from_val`` with ``to_val``.
Replaces the first ``max`` occurrences of the specified element with another
specified element.
See Also:
This is an alias for :py:func:`array_replace_n`.
"""
return array_replace_n(array, from_val, to_val, max)
def array_replace_all(array: Expr, from_val: Expr, to_val: Expr) -> Expr:
"""Replaces all occurrences of ``from_val`` with ``to_val``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 1]]})
>>> result = df.select(
... dfn.functions.array_replace_all(dfn.col("a"), dfn.lit(1),
... dfn.lit(9)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[9, 2, 9]
"""
return Expr(f.array_replace_all(array.expr, from_val.expr, to_val.expr))
def list_replace_all(array: Expr, from_val: Expr, to_val: Expr) -> Expr:
"""Replaces all occurrences of ``from_val`` with ``to_val``.
See Also:
This is an alias for :py:func:`array_replace_all`.
"""
return array_replace_all(array, from_val, to_val)
def array_sort(array: Expr, descending: bool = False, null_first: bool = False) -> Expr:
"""Sort an array.
Args:
array: The input array to sort.
descending: If True, sorts in descending order.
null_first: If True, nulls will be returned at the beginning of the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[3, 1, 2]]})
>>> result = df.select(
... dfn.functions.array_sort(
... dfn.col("a")
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 3]
>>> df = ctx.from_pydict({"a": [[3, None, 1]]})
>>> result = df.select(
... dfn.functions.array_sort(
... dfn.col("a"), descending=True, null_first=True,
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[None, 3, 1]
"""
desc = "DESC" if descending else "ASC"
nulls_first = "NULLS FIRST" if null_first else "NULLS LAST"
return Expr(
f.array_sort(
array.expr,
Expr.literal(pa.scalar(desc, type=pa.string())).expr,
Expr.literal(pa.scalar(nulls_first, type=pa.string())).expr,
)
)
def list_sort(array: Expr, descending: bool = False, null_first: bool = False) -> Expr:
"""Sorts the array.
See Also:
This is an alias for :py:func:`array_sort`.
"""
return array_sort(array, descending=descending, null_first=null_first)
def array_slice(
array: Expr,
begin: Expr | int,
end: Expr | int,
stride: Expr | int | None = None,
) -> Expr:
"""Returns a slice of the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3, 4]]})
>>> result = df.select(
... dfn.functions.array_slice(dfn.col("a"), 2, 3).alias("result"))
>>> result.collect_column("result")[0].as_py()
[2, 3]
Use ``stride`` to skip elements:
>>> result = df.select(
... dfn.functions.array_slice(
... dfn.col("a"), 1, 4, stride=2,
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 3]
"""
begin = coerce_to_expr(begin)
end = coerce_to_expr(end)
stride = coerce_to_expr_or_none(stride)
return Expr(
f.array_slice(
array.expr,
begin.expr,
end.expr,
stride.expr if stride is not None else None,
)
)
def list_slice(
array: Expr, begin: Expr | int, end: Expr | int, stride: Expr | int | None = None
) -> Expr:
"""Returns a slice of the array.
See Also:
This is an alias for :py:func:`array_slice`.
"""
return array_slice(array, begin, end, stride)
def array_intersect(array1: Expr, array2: Expr) -> Expr:
"""Returns the intersection of ``array1`` and ``array2``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]], "b": [[2, 3, 4]]})
>>> result = df.select(
... dfn.functions.array_intersect(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> sorted(
... result.collect_column("result")[0].as_py()
... )
[2, 3]
"""
return Expr(f.array_intersect(array1.expr, array2.expr))
def list_intersect(array1: Expr, array2: Expr) -> Expr:
"""Returns an the intersection of ``array1`` and ``array2``.
See Also:
This is an alias for :py:func:`array_intersect`.
"""
return array_intersect(array1, array2)
def array_union(array1: Expr, array2: Expr) -> Expr:
"""Returns an array of the elements in the union of array1 and array2.
Duplicate rows will not be returned.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]], "b": [[2, 3, 4]]})
>>> result = df.select(
... dfn.functions.array_union(
... dfn.col("a"), dfn.col("b")
... ).alias("result")
... )
>>> sorted(
... result.collect_column("result")[0].as_py()
... )
[1, 2, 3, 4]
"""
return Expr(f.array_union(array1.expr, array2.expr))
def list_union(array1: Expr, array2: Expr) -> Expr:
"""Returns an array of the elements in the union of array1 and array2.
Duplicate rows will not be returned.
See Also:
This is an alias for :py:func:`array_union`.
"""
return array_union(array1, array2)
def array_except(array1: Expr, array2: Expr) -> Expr:
"""Returns the elements that appear in ``array1`` but not in ``array2``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]], "b": [[2, 3, 4]]})
>>> result = df.select(
... dfn.functions.array_except(dfn.col("a"), dfn.col("b")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1]
"""
return Expr(f.array_except(array1.expr, array2.expr))
def list_except(array1: Expr, array2: Expr) -> Expr:
"""Returns the elements that appear in ``array1`` but not in the ``array2``.
See Also:
This is an alias for :py:func:`array_except`.
"""
return array_except(array1, array2)
def array_resize(array: Expr, size: Expr | int, value: Expr) -> Expr:
"""Returns an array with the specified size filled.
If ``size`` is greater than the ``array`` length, the additional entries will
be filled with the given ``value``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2]]})
>>> result = df.select(
... dfn.functions.array_resize(dfn.col("a"), 4, dfn.lit(0)).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 0, 0]
"""
size = coerce_to_expr(size)
return Expr(f.array_resize(array.expr, size.expr, value.expr))
def list_resize(array: Expr, size: Expr | int, value: Expr) -> Expr:
"""Returns an array with the specified size filled.
If ``size`` is greater than the ``array`` length, the additional entries will be
filled with the given ``value``.
See Also:
This is an alias for :py:func:`array_resize`.
"""
return array_resize(array, size, value)
def array_any_value(array: Expr) -> Expr:
"""Returns the first non-null element in the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[None, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_any_value(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
2
"""
return Expr(f.array_any_value(array.expr))
def list_any_value(array: Expr) -> Expr:
"""Returns the first non-null element in the array.
See Also:
This is an alias for :py:func:`array_any_value`.
"""
return array_any_value(array)
def array_distance(array1: Expr, array2: Expr) -> Expr:
"""Returns the Euclidean distance between two numeric arrays.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1.0, 2.0]], "b": [[1.0, 4.0]]})
>>> result = df.select(
... dfn.functions.array_distance(
... dfn.col("a"), dfn.col("b"),
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
2.0
"""
return Expr(f.array_distance(array1.expr, array2.expr))
def list_distance(array1: Expr, array2: Expr) -> Expr:
"""Returns the Euclidean distance between two numeric arrays.
See Also:
This is an alias for :py:func:`array_distance`.
"""
return array_distance(array1, array2)
def array_max(array: Expr) -> Expr:
"""Returns the maximum value in the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_max(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
3
"""
return Expr(f.array_max(array.expr))
def list_max(array: Expr) -> Expr:
"""Returns the maximum value in the array.
See Also:
This is an alias for :py:func:`array_max`.
"""
return array_max(array)
def array_min(array: Expr) -> Expr:
"""Returns the minimum value in the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_min(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
1
"""
return Expr(f.array_min(array.expr))
def list_min(array: Expr) -> Expr:
"""Returns the minimum value in the array.
See Also:
This is an alias for :py:func:`array_min`.
"""
return array_min(array)
def array_reverse(array: Expr) -> Expr:
"""Reverses the order of elements in the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(
... dfn.functions.array_reverse(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[3, 2, 1]
"""
return Expr(f.array_reverse(array.expr))
def list_reverse(array: Expr) -> Expr:
"""Reverses the order of elements in the array.
See Also:
This is an alias for :py:func:`array_reverse`.
"""
return array_reverse(array)
def arrays_zip(*arrays: Expr) -> Expr:
"""Combines multiple arrays into a single array of structs.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2]], "b": [[3, 4]]})
>>> result = df.select(
... dfn.functions.arrays_zip(dfn.col("a"), dfn.col("b")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[{'1': 1, '2': 3}, {'1': 2, '2': 4}]
"""
args = [a.expr for a in arrays]
return Expr(f.arrays_zip(args))
def list_zip(*arrays: Expr) -> Expr:
"""Combines multiple arrays into a single array of structs.
See Also:
This is an alias for :py:func:`arrays_zip`.
"""
return arrays_zip(*arrays)
def string_to_array(
string: Expr, delimiter: Expr | str, null_string: Expr | str | None = None
) -> Expr:
"""Splits a string based on a delimiter and returns an array of parts.
Any parts matching the optional ``null_string`` will be replaced with ``NULL``.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["hello,world"]})
>>> result = df.select(
... dfn.functions.string_to_array(dfn.col("a"), ",").alias("result"))
>>> result.collect_column("result")[0].as_py()
['hello', 'world']
Replace parts matching a ``null_string`` with ``NULL``:
>>> result = df.select(
... dfn.functions.string_to_array(
... dfn.col("a"), ",", null_string="world",
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
['hello', None]
"""
delimiter = coerce_to_expr(delimiter)
null_string = coerce_to_expr_or_none(null_string)
return Expr(
f.string_to_array(
string.expr,
delimiter.expr,
null_string.expr if null_string is not None else None,
)
)
def string_to_list(
string: Expr, delimiter: Expr | str, null_string: Expr | str | None = None
) -> Expr:
"""Splits a string based on a delimiter and returns an array of parts.
See Also:
This is an alias for :py:func:`string_to_array`.
"""
return string_to_array(string, delimiter, null_string)
def gen_series(start: Expr, stop: Expr, step: Expr | None = None) -> Expr:
"""Creates a list of values in the range between start and stop.
Unlike :py:func:`range`, this includes the upper bound.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0]})
>>> result = df.select(
... dfn.functions.gen_series(
... dfn.lit(1), dfn.lit(5),
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 3, 4, 5]
Specify a custom ``step``:
>>> result = df.select(
... dfn.functions.gen_series(
... dfn.lit(1), dfn.lit(10), step=dfn.lit(3),
... ).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 4, 7, 10]
"""
step_expr = step.expr if step is not None else None
return Expr(f.gen_series(start.expr, stop.expr, step_expr))
def generate_series(start: Expr, stop: Expr, step: Expr | None = None) -> Expr:
"""Creates a list of values in the range between start and stop.
Unlike :py:func:`range`, this includes the upper bound.
See Also:
This is an alias for :py:func:`gen_series`.
"""
return gen_series(start, stop, step)
def flatten(array: Expr) -> Expr:
"""Flattens an array of arrays into a single array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[[1, 2], [3, 4]]]})
>>> result = df.select(dfn.functions.flatten(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
[1, 2, 3, 4]
"""
return Expr(f.flatten(array.expr))
def cardinality(array: Expr) -> Expr:
"""Returns the total number of elements in the array.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [[1, 2, 3]]})
>>> result = df.select(dfn.functions.cardinality(dfn.col("a")).alias("result"))
>>> result.collect_column("result")[0].as_py()
3
"""
return Expr(f.cardinality(array.expr))
def empty(array: Expr) -> Expr:
"""Returns true if the array is empty.
See Also:
This is an alias for :py:func:`array_empty`.
"""
return array_empty(array)
# map functions
def make_map(*args: Any) -> Expr:
"""Returns a map expression.
Supports three calling conventions:
- ``make_map({"a": 1, "b": 2})`` — from a Python dictionary.
- ``make_map([keys], [values])`` — from a list of keys and a list of
their associated values. Both lists must be the same length.
- ``make_map(k1, v1, k2, v2, ...)`` — from alternating keys and their
associated values.
Keys and values that are not already :py:class:`~datafusion.expr.Expr`
are automatically converted to literal expressions.
Examples:
From a dictionary:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.make_map({"a": 1, "b": 2}).alias("m"))
>>> result.collect_column("m")[0].as_py()
[('a', 1), ('b', 2)]
From two lists:
>>> df = ctx.from_pydict({"key": ["x", "y"], "val": [10, 20]})
>>> df = df.select(
... dfn.functions.make_map(
... [dfn.col("key")], [dfn.col("val")]
... ).alias("m"))
>>> df.collect_column("m")[0].as_py()
[('x', 10)]
From alternating keys and values:
>>> df = ctx.from_pydict({"a": [1]})
>>> result = df.select(
... dfn.functions.make_map("x", 1, "y", 2).alias("m"))
>>> result.collect_column("m")[0].as_py()
[('x', 1), ('y', 2)]
"""
if len(args) == 1 and isinstance(args[0], dict):
key_list = list(args[0].keys())
value_list = list(args[0].values())
elif (
len(args) == 2 # noqa: PLR2004
and isinstance(args[0], list)
and isinstance(args[1], list)
):
if len(args[0]) != len(args[1]):
msg = "make_map requires key and value lists to be the same length"
raise ValueError(msg)
key_list = args[0]
value_list = args[1]
elif len(args) >= 2 and len(args) % 2 == 0: # noqa: PLR2004
key_list = list(args[0::2])
value_list = list(args[1::2])
else:
msg = (
"make_map expects a dict, two lists, or an even number of "
"key-value arguments"
)
raise ValueError(msg)
key_exprs = [k if isinstance(k, Expr) else Expr.literal(k) for k in key_list]
val_exprs = [v if isinstance(v, Expr) else Expr.literal(v) for v in value_list]
return Expr(f.make_map([k.expr for k in key_exprs], [v.expr for v in val_exprs]))
def map_keys(map: Expr) -> Expr:
"""Returns a list of all keys in the map.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> df = df.select(
... dfn.functions.make_map({"x": 1, "y": 2}).alias("m"))
>>> result = df.select(
... dfn.functions.map_keys(dfn.col("m")).alias("keys"))
>>> result.collect_column("keys")[0].as_py()
['x', 'y']
"""
return Expr(f.map_keys(map.expr))
def map_values(map: Expr) -> Expr:
"""Returns a list of all values in the map.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> df = df.select(
... dfn.functions.make_map({"x": 1, "y": 2}).alias("m"))
>>> result = df.select(
... dfn.functions.map_values(dfn.col("m")).alias("vals"))
>>> result.collect_column("vals")[0].as_py()
[1, 2]
"""
return Expr(f.map_values(map.expr))
def map_extract(map: Expr, key: Expr) -> Expr:
"""Returns the value for a given key in the map.
Returns ``[None]`` if the key is absent.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> df = df.select(
... dfn.functions.make_map({"x": 1, "y": 2}).alias("m"))
>>> result = df.select(
... dfn.functions.map_extract(
... dfn.col("m"), dfn.lit("x")
... ).alias("val"))
>>> result.collect_column("val")[0].as_py()
[1]
"""
return Expr(f.map_extract(map.expr, key.expr))
def map_entries(map: Expr) -> Expr:
"""Returns a list of all entries (key-value struct pairs) in the map.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1]})
>>> df = df.select(
... dfn.functions.make_map({"x": 1, "y": 2}).alias("m"))
>>> result = df.select(
... dfn.functions.map_entries(dfn.col("m")).alias("entries"))
>>> result.collect_column("entries")[0].as_py()
[{'key': 'x', 'value': 1}, {'key': 'y', 'value': 2}]
"""
return Expr(f.map_entries(map.expr))
def element_at(map: Expr, key: Expr) -> Expr:
"""Returns the value for a given key in the map.
Returns ``[None]`` if the key is absent.
See Also:
This is an alias for :py:func:`map_extract`.
"""
return map_extract(map, key)
# aggregate functions
def approx_distinct(
expression: Expr,
filter: Expr | None = None,
) -> Expr:
"""Returns the approximate number of distinct values.
This aggregate function is similar to :py:func:`count` with distinct set, but it
will approximate the number of distinct entries. It may return significantly faster
than :py:func:`count` for some DataFrames.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: Values to check for distinct entries
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.approx_distinct(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py() == 3
True
>>> result = df.aggregate(
... [], [dfn.functions.approx_distinct(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py() == 2
True
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.approx_distinct(expression.expr, filter=filter_raw))
def approx_median(expression: Expr, filter: Expr | None = None) -> Expr:
"""Returns the approximate median value.
This aggregate function is similar to :py:func:`median`, but it will only
approximate the median. It may return significantly faster for some DataFrames.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``, and ``distinct``.
Args:
expression: Values to find the median for
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.approx_median(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.approx_median(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.approx_median(expression.expr, filter=filter_raw))
def approx_percentile_cont(
sort_expression: Expr | SortExpr,
percentile: float,
num_centroids: int | None = None,
filter: Expr | None = None,
) -> Expr:
"""Returns the value that is approximately at a given percentile of ``expr``.
This aggregate function assumes the input values form a continuous distribution.
Suppose you have a DataFrame which consists of 100 different test scores. If you
called this function with a percentile of 0.9, it would return the value of the
test score that is above 90% of the other test scores. The returned value may be
between two of the values.
This function uses the [t-digest](https://arxiv.org/abs/1902.04023) algorithm to
compute the percentile. You can limit the number of bins used in this algorithm by
setting the ``num_centroids`` parameter.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
sort_expression: Values for which to find the approximate percentile
percentile: This must be between 0.0 and 1.0, inclusive
num_centroids: Max bin size for the t-digest algorithm
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0, 4.0, 5.0]})
>>> result = df.aggregate(
... [], [dfn.functions.approx_percentile_cont(
... dfn.col("a"), 0.5
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3.0
>>> result = df.aggregate(
... [], [dfn.functions.approx_percentile_cont(
... dfn.col("a"), 0.5,
... num_centroids=10,
... filter=dfn.col("a") > dfn.lit(1.0),
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3.5
"""
sort_expr_raw = sort_or_default(sort_expression)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.approx_percentile_cont(
sort_expr_raw, percentile, num_centroids=num_centroids, filter=filter_raw
)
)
def approx_percentile_cont_with_weight(
sort_expression: Expr | SortExpr,
weight: Expr,
percentile: float,
num_centroids: int | None = None,
filter: Expr | None = None,
) -> Expr:
"""Returns the value of the weighted approximate percentile.
This aggregate function is similar to :py:func:`approx_percentile_cont` except that
it uses the associated associated weights.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
sort_expression: Values for which to find the approximate percentile
weight: Relative weight for each of the values in ``expression``
percentile: This must be between 0.0 and 1.0, inclusive
num_centroids: Max bin size for the t-digest algorithm
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0], "w": [1.0, 1.0, 1.0]})
>>> result = df.aggregate(
... [], [dfn.functions.approx_percentile_cont_with_weight(
... dfn.col("a"), dfn.col("w"), 0.5
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.approx_percentile_cont_with_weight(
... dfn.col("a"), dfn.col("w"), 0.5,
... num_centroids=10,
... filter=dfn.col("a") > dfn.lit(1.0),
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.5
"""
sort_expr_raw = sort_or_default(sort_expression)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.approx_percentile_cont_with_weight(
sort_expr_raw,
weight.expr,
percentile,
num_centroids=num_centroids,
filter=filter_raw,
)
)
def percentile_cont(
sort_expression: Expr | SortExpr,
percentile: float,
filter: Expr | None = None,
) -> Expr:
"""Computes the exact percentile of input values using continuous interpolation.
Unlike :py:func:`approx_percentile_cont`, this function computes the exact
percentile value rather than an approximation.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
sort_expression: Values for which to find the percentile
percentile: This must be between 0.0 and 1.0, inclusive
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0, 4.0, 5.0]})
>>> result = df.aggregate(
... [], [dfn.functions.percentile_cont(
... dfn.col("a"), 0.5
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3.0
>>> result = df.aggregate(
... [], [dfn.functions.percentile_cont(
... dfn.col("a"), 0.5,
... filter=dfn.col("a") > dfn.lit(1.0),
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3.5
"""
sort_expr_raw = sort_or_default(sort_expression)
filter_raw = filter.expr if filter is not None else None
return Expr(f.percentile_cont(sort_expr_raw, percentile, filter=filter_raw))
def quantile_cont(
sort_expression: Expr | SortExpr,
percentile: float,
filter: Expr | None = None,
) -> Expr:
"""Computes the exact percentile of input values using continuous interpolation.
See Also:
This is an alias for :py:func:`percentile_cont`.
"""
return percentile_cont(sort_expression, percentile, filter)
def array_agg(
expression: Expr,
distinct: bool = False,
filter: Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Aggregate values into an array.
Currently ``distinct`` and ``order_by`` cannot be used together. As a work around,
consider :py:func:`array_sort` after aggregation.
[Issue Tracker](https://github.com/apache/datafusion/issues/12371)
If using the builder functions described in ref:`_aggregation` this function ignores
the option ``null_treatment``.
Args:
expression: Values to combine into an array
distinct: If True, a single entry for each distinct value will be in the result
filter: If provided, only compute against rows for which the filter is True
order_by: Order the resultant array values. Accepts column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.array_agg(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
[1, 2, 3]
>>> df = ctx.from_pydict({"a": [3, 1, 2, 1]})
>>> result = df.aggregate(
... [], [dfn.functions.array_agg(
... dfn.col("a"), distinct=True,
... ).alias("v")])
>>> sorted(result.collect_column("v")[0].as_py())
[1, 2, 3]
>>> result = df.aggregate(
... [], [dfn.functions.array_agg(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1),
... order_by="a",
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
[2, 3]
"""
order_by_raw = sort_list_to_raw_sort_list(order_by)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.array_agg(
expression.expr, distinct=distinct, filter=filter_raw, order_by=order_by_raw
)
)
def grouping(
expression: Expr,
distinct: bool = False,
filter: Expr | None = None,
) -> Expr:
"""Indicates whether a column is aggregated across in the current row.
Returns 0 when the column is part of the grouping key for that row
(i.e., the row contains per-group results for that column). Returns 1
when the column is *not* part of the grouping key (i.e., the row's
aggregate spans all values of that column).
This function is meaningful with
:py:meth:`GroupingSet.rollup <datafusion.expr.GroupingSet.rollup>`,
:py:meth:`GroupingSet.cube <datafusion.expr.GroupingSet.cube>`, or
:py:meth:`GroupingSet.grouping_sets <datafusion.expr.GroupingSet.grouping_sets>`,
where different rows are grouped by different subsets of columns. In a
default aggregation without grouping sets every column is always part
of the key, so ``grouping()`` always returns 0.
.. warning::
Due to an upstream DataFusion limitation
(`#21411 <https://github.com/apache/datafusion/issues/21411>`_),
``.alias()`` cannot be applied directly to a ``grouping()``
expression. Doing so will raise an error at execution time. To
rename the column, use
:py:meth:`~datafusion.dataframe.DataFrame.with_column_renamed`
on the result DataFrame instead.
Args:
expression: The column to check grouping status for
distinct: If True, compute on distinct values only
filter: If provided, only compute against rows for which the filter is True
Examples:
With :py:meth:`~datafusion.expr.GroupingSet.rollup`, the result
includes both per-group rows (``grouping(a) = 0``) and a
grand-total row where ``a`` is aggregated across
(``grouping(a) = 1``):
>>> from datafusion.expr import GroupingSet
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 1, 2], "b": [10, 20, 30]})
>>> result = df.aggregate(
... [GroupingSet.rollup(dfn.col("a"))],
... [dfn.functions.sum(dfn.col("b")).alias("s"),
... dfn.functions.grouping(dfn.col("a"))],
... ).sort(dfn.col("a").sort(nulls_first=False))
>>> result.collect_column("s").to_pylist()
[30, 30, 60]
See Also:
:py:class:`~datafusion.expr.GroupingSet`
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.grouping(expression.expr, distinct=distinct, filter=filter_raw))
def avg(
expression: Expr,
distinct: bool = False,
filter: Expr | None = None,
) -> Expr:
"""Returns the average value.
This aggregate function expects a numeric expression and will return a float.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``.
Args:
expression: Values to combine into an array
distinct: If True, duplicate values are removed before averaging.
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.avg(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.avg(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.5
>>> df = ctx.from_pydict({"a": [1.0, 1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.avg(
... dfn.col("a"), distinct=True,
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.avg(expression.expr, distinct=distinct, filter=filter_raw))
def corr(value_y: Expr, value_x: Expr, filter: Expr | None = None) -> Expr:
"""Returns the correlation coefficient between ``value1`` and ``value2``.
This aggregate function expects both values to be numeric and will return a float.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
value_y: The dependent variable for correlation
value_x: The independent variable for correlation
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0], "b": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.corr(
... dfn.col("a"), dfn.col("b")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
>>> result = df.aggregate(
... [], [dfn.functions.corr(
... dfn.col("a"), dfn.col("b"),
... filter=dfn.col("a") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.corr(value_y.expr, value_x.expr, filter=filter_raw))
def count(
expressions: Expr | list[Expr] | None = None,
distinct: bool = False,
filter: Expr | None = None,
) -> Expr:
"""Returns the number of rows that match the given arguments.
This aggregate function will count the non-null rows provided in the expression.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``.
Args:
expressions: Argument to perform bitwise calculation on
distinct: If True, a single entry for each distinct value will be in the result
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.count(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3
>>> df = ctx.from_pydict({"a": [1, 1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.count(
... dfn.col("a"), distinct=True,
... filter=dfn.col("a") > dfn.lit(1),
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2
"""
filter_raw = filter.expr if filter is not None else None
if expressions is None:
args = [Expr.literal(1).expr]
elif isinstance(expressions, list):
args = [arg.expr for arg in expressions]
else:
args = [expressions.expr]
return Expr(f.count(*args, distinct=distinct, filter=filter_raw))
def covar_pop(value_y: Expr, value_x: Expr, filter: Expr | None = None) -> Expr:
"""Computes the population covariance.
This aggregate function expects both values to be numeric and will return a float.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
value_y: The dependent variable for covariance
value_x: The independent variable for covariance
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 5.0, 10.0], "b": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [],
... [dfn.functions.covar_pop(
... dfn.col("a"), dfn.col("b")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
3.0
>>> df = ctx.from_pydict(
... {"a": [0.0, 1.0, 3.0], "b": [0.0, 1.0, 3.0]})
>>> result = df.aggregate(
... [],
... [dfn.functions.covar_pop(
... dfn.col("a"), dfn.col("b"),
... filter=dfn.col("a") > dfn.lit(0.0)
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
1.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.covar_pop(value_y.expr, value_x.expr, filter=filter_raw))
def covar_samp(value_y: Expr, value_x: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample covariance.
This aggregate function expects both values to be numeric and will return a float.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
value_y: The dependent variable for covariance
value_x: The independent variable for covariance
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0], "b": [4.0, 5.0, 6.0]})
>>> result = df.aggregate(
... [], [dfn.functions.covar_samp(
... dfn.col("a"), dfn.col("b")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
>>> result = df.aggregate(
... [], [dfn.functions.covar_samp(
... dfn.col("a"), dfn.col("b"),
... filter=dfn.col("a") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.covar_samp(value_y.expr, value_x.expr, filter=filter_raw))
def covar(value_y: Expr, value_x: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample covariance.
See Also:
This is an alias for :py:func:`covar_samp`.
"""
return covar_samp(value_y, value_x, filter)
def max(expression: Expr, filter: Expr | None = None) -> Expr:
"""Aggregate function that returns the maximum value of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The value to find the maximum of
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.max(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3
>>> result = df.aggregate(
... [], [dfn.functions.max(
... dfn.col("a"),
... filter=dfn.col("a") < dfn.lit(3)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.max(expression.expr, filter=filter_raw))
def mean(expression: Expr, filter: Expr | None = None) -> Expr:
"""Returns the average (mean) value of the argument.
See Also:
This is an alias for :py:func:`avg`.
"""
return avg(expression, filter)
def median(
expression: Expr, distinct: bool = False, filter: Expr | None = None
) -> Expr:
"""Computes the median of a set of numbers.
This aggregate function returns the median value of the expression for the given
aggregate function.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``.
Args:
expression: The value to compute the median of
distinct: If True, a single entry for each distinct value will be in the result
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.median(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> df = ctx.from_pydict({"a": [1.0, 1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.median(
... dfn.col("a"), distinct=True,
... filter=dfn.col("a") < dfn.lit(3.0),
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.median(expression.expr, distinct=distinct, filter=filter_raw))
def min(expression: Expr, filter: Expr | None = None) -> Expr:
"""Aggregate function that returns the minimum value of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The value to find the minimum of
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.min(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1
>>> result = df.aggregate(
... [], [dfn.functions.min(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.min(expression.expr, filter=filter_raw))
def sum(
expression: Expr,
distinct: bool = False,
filter: Expr | None = None,
) -> Expr:
"""Computes the sum of a set of numbers.
This aggregate function expects a numeric expression.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``.
Args:
expression: Values to combine into an array
distinct: If True, duplicate values are removed before summing.
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.sum(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
6
>>> result = df.aggregate(
... [], [dfn.functions.sum(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
5
>>> df = ctx.from_pydict({"a": [1, 1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.sum(
... dfn.col("a"), distinct=True,
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
6
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.sum(expression.expr, distinct=distinct, filter=filter_raw))
def stddev(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the standard deviation of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The value to find the minimum of
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [2.0, 4.0, 6.0]})
>>> result = df.aggregate(
... [], [dfn.functions.stddev(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.stddev(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(2.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.41...
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.stddev(expression.expr, filter=filter_raw))
def stddev_pop(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the population standard deviation of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The value to find the minimum of
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [0.0, 1.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.stddev_pop(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
1.247...
>>> df = ctx.from_pydict({"a": [0.0, 1.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.stddev_pop(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(0.0)
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
1.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.stddev_pop(expression.expr, filter=filter_raw))
def stddev_samp(arg: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample standard deviation of the argument.
See Also:
This is an alias for :py:func:`stddev`.
"""
return stddev(arg, filter=filter)
def var(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample variance of the argument.
See Also:
This is an alias for :py:func:`var_samp`.
"""
return var_samp(expression, filter)
def var_pop(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the population variance of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The variable to compute the variance for
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [-1.0, 0.0, 2.0]})
>>> result = df.aggregate(
... [], [dfn.functions.var_pop(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.555...
>>> result = df.aggregate(
... [], [dfn.functions.var_pop(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(-1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.var_pop(expression.expr, filter=filter_raw))
def var_population(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the population variance of the argument.
See Also:
This is an alias for :py:func:`var_pop`.
"""
return var_pop(expression, filter)
def var_samp(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample variance of the argument.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: The variable to compute the variance for
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.var_samp(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
>>> result = df.aggregate(
... [], [dfn.functions.var_samp(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.var_sample(expression.expr, filter=filter_raw))
def var_sample(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the sample variance of the argument.
See Also:
This is an alias for :py:func:`var_samp`.
"""
return var_samp(expression, filter)
def regr_avgx(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the average of the independent variable ``x``.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [4.0, 5.0, 6.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_avgx(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
5.0
>>> result = df.aggregate(
... [], [dfn.functions.regr_avgx(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
5.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_avgx(y.expr, x.expr, filter=filter_raw))
def regr_avgy(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the average of the dependent variable ``y``.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [4.0, 5.0, 6.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_avgy(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.regr_avgy(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_avgy(y.expr, x.expr, filter=filter_raw))
def regr_count(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Counts the number of rows in which both expressions are not null.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [4.0, 5.0, 6.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_count(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3
>>> result = df.aggregate(
... [], [dfn.functions.regr_count(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_count(y.expr, x.expr, filter=filter_raw))
def regr_intercept(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the intercept from the linear regression.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [2.0, 4.0, 6.0], "x": [4.0, 16.0, 36.0]})
>>> result = df.aggregate(
... [],
... [dfn.functions.regr_intercept(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.714...
>>> result = df.aggregate(
... [],
... [dfn.functions.regr_intercept(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(2.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.4
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_intercept(y.expr, x.expr, filter=filter_raw))
def regr_r2(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the R-squared value from linear regression.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [2.0, 4.0, 6.0], "x": [4.0, 16.0, 36.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_r2(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.9795...
>>> result = df.aggregate(
... [], [dfn.functions.regr_r2(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(2.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
1.0
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_r2(y.expr, x.expr, filter=filter_raw))
def regr_slope(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the slope from linear regression.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [2.0, 4.0, 6.0], "x": [4.0, 16.0, 36.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_slope(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.122...
>>> result = df.aggregate(
... [], [dfn.functions.regr_slope(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(2.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.1
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_slope(y.expr, x.expr, filter=filter_raw))
def regr_sxx(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the sum of squares of the independent variable ``x``.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_sxx(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.regr_sxx(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_sxx(y.expr, x.expr, filter=filter_raw))
def regr_sxy(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the sum of products of pairs of numbers.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_sxy(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.regr_sxy(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_sxy(y.expr, x.expr, filter=filter_raw))
def regr_syy(
y: Expr,
x: Expr,
filter: Expr | None = None,
) -> Expr:
"""Computes the sum of squares of the dependent variable ``y``.
This is a linear regression aggregate function. Only non-null pairs of the inputs
are evaluated.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
y: The linear regression dependent variable
x: The linear regression independent variable
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"y": [1.0, 2.0, 3.0], "x": [1.0, 2.0, 3.0]})
>>> result = df.aggregate(
... [], [dfn.functions.regr_syy(
... dfn.col("y"), dfn.col("x")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
2.0
>>> result = df.aggregate(
... [], [dfn.functions.regr_syy(
... dfn.col("y"), dfn.col("x"),
... filter=dfn.col("y") > dfn.lit(1.0)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
0.5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.regr_syy(y.expr, x.expr, filter=filter_raw))
def first_value(
expression: Expr,
filter: Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
null_treatment: NullTreatment = NullTreatment.RESPECT_NULLS,
) -> Expr:
"""Returns the first value in a group of values.
This aggregate function will return the first value in the partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the option ``distinct``.
Args:
expression: Argument to perform bitwise calculation on
filter: If provided, only compute against rows for which the filter is True
order_by: Set the ordering of the expression to evaluate. Accepts
column names or expressions.
null_treatment: Assign whether to respect or ignore null values.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30]})
>>> result = df.aggregate(
... [], [dfn.functions.first_value(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
10
>>> df = ctx.from_pydict({"a": [None, 20, 10]})
>>> result = df.aggregate(
... [], [dfn.functions.first_value(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(10),
... order_by="a",
... null_treatment=dfn.common.NullTreatment.IGNORE_NULLS,
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
20
"""
order_by_raw = sort_list_to_raw_sort_list(order_by)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.first_value(
expression.expr,
filter=filter_raw,
order_by=order_by_raw,
null_treatment=null_treatment.value,
)
)
def last_value(
expression: Expr,
filter: Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
null_treatment: NullTreatment = NullTreatment.RESPECT_NULLS,
) -> Expr:
"""Returns the last value in a group of values.
This aggregate function will return the last value in the partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the option ``distinct``.
Args:
expression: Argument to perform bitwise calculation on
filter: If provided, only compute against rows for which the filter is True
order_by: Set the ordering of the expression to evaluate. Accepts
column names or expressions.
null_treatment: Assign whether to respect or ignore null values.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30]})
>>> result = df.aggregate(
... [], [dfn.functions.last_value(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
30
>>> df = ctx.from_pydict({"a": [None, 20, 10]})
>>> result = df.aggregate(
... [], [dfn.functions.last_value(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(10),
... order_by="a",
... null_treatment=dfn.common.NullTreatment.IGNORE_NULLS,
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
20
"""
order_by_raw = sort_list_to_raw_sort_list(order_by)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.last_value(
expression.expr,
filter=filter_raw,
order_by=order_by_raw,
null_treatment=null_treatment.value,
)
)
def nth_value(
expression: Expr,
n: int,
filter: Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
null_treatment: NullTreatment = NullTreatment.RESPECT_NULLS,
) -> Expr:
"""Returns the n-th value in a group of values.
This aggregate function will return the n-th value in the partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the option ``distinct``.
Args:
expression: Argument to perform bitwise calculation on
n: Index of value to return. Starts at 1.
filter: If provided, only compute against rows for which the filter is True
order_by: Set the ordering of the expression to evaluate. Accepts
column names or expressions.
null_treatment: Assign whether to respect or ignore null values.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30]})
>>> result = df.aggregate(
... [], [dfn.functions.nth_value(
... dfn.col("a"), 1
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
10
>>> result = df.aggregate(
... [], [dfn.functions.nth_value(
... dfn.col("a"), 1,
... filter=dfn.col("a") > dfn.lit(10),
... order_by="a",
... null_treatment=dfn.common.NullTreatment.IGNORE_NULLS,
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
20
"""
order_by_raw = sort_list_to_raw_sort_list(order_by)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.nth_value(
expression.expr,
n,
filter=filter_raw,
order_by=order_by_raw,
null_treatment=null_treatment.value,
)
)
def bit_and(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the bitwise AND of the argument.
This aggregate function will bitwise compare every value in the input partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: Argument to perform bitwise calculation on
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [7, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_and(
... dfn.col("a")
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
3
>>> df = ctx.from_pydict({"a": [7, 5, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_and(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(3)
... ).alias("v")])
>>> result.collect_column("v")[0].as_py()
5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.bit_and(expression.expr, filter=filter_raw))
def bit_or(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the bitwise OR of the argument.
This aggregate function will bitwise compare every value in the input partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: Argument to perform bitwise calculation on
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_or(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
3
>>> df = ctx.from_pydict({"a": [1, 2, 4]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_or(
... dfn.col("a"),
... filter=dfn.col("a") > dfn.lit(1)
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
6
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.bit_or(expression.expr, filter=filter_raw))
def bit_xor(
expression: Expr, distinct: bool = False, filter: Expr | None = None
) -> Expr:
"""Computes the bitwise XOR of the argument.
This aggregate function will bitwise compare every value in the input partition.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by`` and ``null_treatment``.
Args:
expression: Argument to perform bitwise calculation on
distinct: If True, evaluate each unique value of expression only once
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [5, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_xor(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
6
>>> df = ctx.from_pydict({"a": [5, 5, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bit_xor(
... dfn.col("a"), distinct=True,
... filter=dfn.col("a") > dfn.lit(3),
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
5
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.bit_xor(expression.expr, distinct=distinct, filter=filter_raw))
def bool_and(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the boolean AND of the argument.
This aggregate function will compare every value in the input partition. These are
expected to be boolean values.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: Argument to perform calculation on
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [True, True, False]})
>>> result = df.aggregate(
... [], [dfn.functions.bool_and(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
False
>>> df = ctx.from_pydict(
... {"a": [True, True, False], "b": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bool_and(
... dfn.col("a"),
... filter=dfn.col("b") < dfn.lit(3)
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
True
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.bool_and(expression.expr, filter=filter_raw))
def bool_or(expression: Expr, filter: Expr | None = None) -> Expr:
"""Computes the boolean OR of the argument.
This aggregate function will compare every value in the input partition. These are
expected to be boolean values.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``order_by``, ``null_treatment``, and ``distinct``.
Args:
expression: Argument to perform calculation on
filter: If provided, only compute against rows for which the filter is True
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [False, False, True]})
>>> result = df.aggregate(
... [], [dfn.functions.bool_or(
... dfn.col("a")
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
True
>>> df = ctx.from_pydict(
... {"a": [False, False, True], "b": [1, 2, 3]})
>>> result = df.aggregate(
... [], [dfn.functions.bool_or(
... dfn.col("a"),
... filter=dfn.col("b") < dfn.lit(3)
... ).alias("v")]
... )
>>> result.collect_column("v")[0].as_py()
False
"""
filter_raw = filter.expr if filter is not None else None
return Expr(f.bool_or(expression.expr, filter=filter_raw))
def lead(
arg: Expr,
shift_offset: int = 1,
default_value: Any | None = None,
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a lead window function.
Lead operation will return the argument that is in the next shift_offset-th row in
the partition. For example ``lead(col("b"), shift_offset=3, default_value=5)`` will
return the 3rd following value in column ``b``. At the end of the partition, where
no further values can be returned it will return the default value of 5.
Here is an example of both the ``lead`` and :py:func:`datafusion.functions.lag`
functions on a simple DataFrame::
+--------+------+-----+
| points | lead | lag |
+--------+------+-----+
| 100 | 100 | |
| 100 | 50 | 100 |
| 50 | 25 | 100 |
| 25 | | 50 |
+--------+------+-----+
To set window function parameters use the window builder approach described in the
ref:`_window_functions` online documentation.
Args:
arg: Value to return
shift_offset: Number of rows following the current row.
default_value: Value to return if shift_offet row does not exist.
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.lead(
... dfn.col("a"), shift_offset=1,
... default_value=0, order_by="a"
... ).alias("lead"))
>>> result.sort(dfn.col("a")).collect_column("lead").to_pylist()
[2, 3, 0]
>>> df = ctx.from_pydict({"g": ["a", "a", "b"], "v": [1, 2, 3]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.lead(
... dfn.col("v"), shift_offset=1, default_value=0,
... partition_by=dfn.col("g"), order_by="v",
... ).alias("lead"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("lead").to_pylist()
[2, 0, 0]
"""
if not isinstance(default_value, pa.Scalar) and default_value is not None:
default_value = pa.scalar(default_value)
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.lead(
arg.expr,
shift_offset,
default_value,
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def lag(
arg: Expr,
shift_offset: int = 1,
default_value: Any | None = None,
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a lag window function.
Lag operation will return the argument that is in the previous shift_offset-th row
in the partition. For example ``lag(col("b"), shift_offset=3, default_value=5)``
will return the 3rd previous value in column ``b``. At the beginning of the
partition, where no values can be returned it will return the default value of 5.
Here is an example of both the ``lag`` and :py:func:`datafusion.functions.lead`
functions on a simple DataFrame::
+--------+------+-----+
| points | lead | lag |
+--------+------+-----+
| 100 | 100 | |
| 100 | 50 | 100 |
| 50 | 25 | 100 |
| 25 | | 50 |
+--------+------+-----+
Args:
arg: Value to return
shift_offset: Number of rows before the current row.
default_value: Value to return if shift_offet row does not exist.
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1, 2, 3]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.lag(
... dfn.col("a"), shift_offset=1,
... default_value=0, order_by="a"
... ).alias("lag"))
>>> result.sort(dfn.col("a")).collect_column("lag").to_pylist()
[0, 1, 2]
>>> df = ctx.from_pydict({"g": ["a", "a", "b"], "v": [1, 2, 3]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.lag(
... dfn.col("v"), shift_offset=1, default_value=0,
... partition_by=dfn.col("g"), order_by="v",
... ).alias("lag"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("lag").to_pylist()
[0, 1, 0]
"""
if not isinstance(default_value, pa.Scalar):
default_value = pa.scalar(default_value)
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.lag(
arg.expr,
shift_offset,
default_value,
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def row_number(
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a row number window function.
Returns the row number of the window function.
Here is an example of the ``row_number`` on a simple DataFrame::
+--------+------------+
| points | row number |
+--------+------------+
| 100 | 1 |
| 100 | 2 |
| 50 | 3 |
| 25 | 4 |
+--------+------------+
Args:
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.row_number(
... order_by="a"
... ).alias("rn"))
>>> result.sort(dfn.col("a")).collect_column("rn").to_pylist()
[1, 2, 3]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "b", "b"], "v": [1, 2, 3, 4]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.row_number(
... partition_by=dfn.col("g"), order_by="v",
... ).alias("rn"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("rn").to_pylist()
[1, 2, 1, 2]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.row_number(
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def rank(
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a rank window function.
Returns the rank based upon the window order. Consecutive equal values will receive
the same rank, but the next different value will not be consecutive but rather the
number of rows that precede it plus one. This is similar to Olympic medals. If two
people tie for gold, the next place is bronze. There would be no silver medal. Here
is an example of a dataframe with a window ordered by descending ``points`` and the
associated rank.
You should set ``order_by`` to produce meaningful results::
+--------+------+
| points | rank |
+--------+------+
| 100 | 1 |
| 100 | 1 |
| 50 | 3 |
| 25 | 4 |
+--------+------+
Args:
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 10, 20]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.rank(
... order_by="a"
... ).alias("rnk")
... )
>>> result.sort(dfn.col("a")).collect_column("rnk").to_pylist()
[1, 1, 3]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "b", "b"], "v": [1, 1, 2, 3]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.rank(
... partition_by=dfn.col("g"), order_by="v",
... ).alias("rnk"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("rnk").to_pylist()
[1, 1, 1, 2]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.rank(
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def dense_rank(
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a dense_rank window function.
This window function is similar to :py:func:`rank` except that the returned values
will be consecutive. Here is an example of a dataframe with a window ordered by
descending ``points`` and the associated dense rank::
+--------+------------+
| points | dense_rank |
+--------+------------+
| 100 | 1 |
| 100 | 1 |
| 50 | 2 |
| 25 | 3 |
+--------+------------+
Args:
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 10, 20]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.dense_rank(
... order_by="a"
... ).alias("dr"))
>>> result.sort(dfn.col("a")).collect_column("dr").to_pylist()
[1, 1, 2]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "b", "b"], "v": [1, 1, 2, 3]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.dense_rank(
... partition_by=dfn.col("g"), order_by="v",
... ).alias("dr"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("dr").to_pylist()
[1, 1, 1, 2]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.dense_rank(
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def percent_rank(
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a percent_rank window function.
This window function is similar to :py:func:`rank` except that the returned values
are the percentage from 0.0 to 1.0 from first to last. Here is an example of a
dataframe with a window ordered by descending ``points`` and the associated percent
rank::
+--------+--------------+
| points | percent_rank |
+--------+--------------+
| 100 | 0.0 |
| 100 | 0.0 |
| 50 | 0.666667 |
| 25 | 1.0 |
+--------+--------------+
Args:
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.percent_rank(
... order_by="a"
... ).alias("pr"))
>>> result.sort(dfn.col("a")).collect_column("pr").to_pylist()
[0.0, 0.5, 1.0]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "a", "b", "b"], "v": [1, 2, 3, 4, 5]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.percent_rank(
... partition_by=dfn.col("g"), order_by="v",
... ).alias("pr"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("pr").to_pylist()
[0.0, 0.5, 1.0, 0.0, 1.0]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.percent_rank(
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def cume_dist(
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a cumulative distribution window function.
This window function is similar to :py:func:`rank` except that the returned values
are the ratio of the row number to the total number of rows. Here is an example of a
dataframe with a window ordered by descending ``points`` and the associated
cumulative distribution::
+--------+-----------+
| points | cume_dist |
+--------+-----------+
| 100 | 0.5 |
| 100 | 0.5 |
| 50 | 0.75 |
| 25 | 1.0 |
+--------+-----------+
Args:
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [1., 2., 2., 3.]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.cume_dist(
... order_by="a"
... ).alias("cd")
... )
>>> result.collect_column("cd").to_pylist()
[0.25..., 0.75..., 0.75..., 1.0...]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "b", "b"], "v": [1, 2, 3, 4]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.cume_dist(
... partition_by=dfn.col("g"), order_by="v",
... ).alias("cd"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("cd").to_pylist()
[0.5, 1.0, 0.5, 1.0]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.cume_dist(
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def ntile(
groups: int,
partition_by: list[Expr] | Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Create a n-tile window function.
This window function orders the window frame into a give number of groups based on
the ordering criteria. It then returns which group the current row is assigned to.
Here is an example of a dataframe with a window ordered by descending ``points``
and the associated n-tile function::
+--------+-------+
| points | ntile |
+--------+-------+
| 120 | 1 |
| 100 | 1 |
| 80 | 2 |
| 60 | 2 |
| 40 | 3 |
| 20 | 3 |
+--------+-------+
Args:
groups: Number of groups for the n-tile to be divided into.
partition_by: Expressions to partition the window frame on.
order_by: Set ordering within the window frame. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": [10, 20, 30, 40]})
>>> result = df.select(
... dfn.col("a"),
... dfn.functions.ntile(
... 2, order_by="a"
... ).alias("nt"))
>>> result.sort(dfn.col("a")).collect_column("nt").to_pylist()
[1, 1, 2, 2]
>>> df = ctx.from_pydict(
... {"g": ["a", "a", "b", "b"], "v": [1, 2, 3, 4]})
>>> result = df.select(
... dfn.col("g"), dfn.col("v"),
... dfn.functions.ntile(
... 2, partition_by=dfn.col("g"), order_by="v",
... ).alias("nt"))
>>> result.sort(dfn.col("g"), dfn.col("v")).collect_column("nt").to_pylist()
[1, 2, 1, 2]
"""
partition_by_raw = expr_list_to_raw_expr_list(partition_by)
order_by_raw = sort_list_to_raw_sort_list(order_by)
return Expr(
f.ntile(
Expr.literal(groups).expr,
partition_by=partition_by_raw,
order_by=order_by_raw,
)
)
def string_agg(
expression: Expr,
delimiter: str,
filter: Expr | None = None,
order_by: list[SortKey] | SortKey | None = None,
) -> Expr:
"""Concatenates the input strings.
This aggregate function will concatenate input strings, ignoring null values, and
separating them with the specified delimiter. Non-string values will be converted to
their string equivalents.
If using the builder functions described in ref:`_aggregation` this function ignores
the options ``distinct`` and ``null_treatment``.
Args:
expression: Argument to perform bitwise calculation on
delimiter: Text to place between each value of expression
filter: If provided, only compute against rows for which the filter is True
order_by: Set the ordering of the expression to evaluate. Accepts
column names or expressions.
Examples:
>>> ctx = dfn.SessionContext()
>>> df = ctx.from_pydict({"a": ["x", "y", "z"]})
>>> result = df.aggregate(
... [], [dfn.functions.string_agg(
... dfn.col("a"), ",", order_by="a"
... ).alias("s")])
>>> result.collect_column("s")[0].as_py()
'x,y,z'
>>> result = df.aggregate(
... [], [dfn.functions.string_agg(
... dfn.col("a"), ",",
... filter=dfn.col("a") > dfn.lit("x"),
... order_by="a",
... ).alias("s")])
>>> result.collect_column("s")[0].as_py()
'y,z'
"""
order_by_raw = sort_list_to_raw_sort_list(order_by)
filter_raw = filter.expr if filter is not None else None
return Expr(
f.string_agg(
expression.expr,
delimiter,
filter=filter_raw,
order_by=order_by_raw,
)
)