sliding_window(timeColumn, windowDuration, slideDuration)

Описание

Функция sliding_window() создает скользящее окно для агрегации данных.

Параметры

• timeColumn: Column - столбец с временными метками
• windowDuration: str - длительность окна (например, "5 minutes", "1 hour", "1 day")
• slideDuration: str - шаг скольжения (например, "1 minute", "15 minutes", "1 hour")

Возвращаемое значение

Column - метка времени начала окна

Пример использования

from pyspark.sql.functions import sliding_window, count, col
from pyspark.sql import SparkSession

# Создаем SparkSession
spark = SparkSession.builder.appName("sliding_window_example").getOrCreate()

# Создаем DataFrame с временными метками
data = [
    ("2023-01-01 10:00:00", 1),
    ("2023-01-01 10:05:00", 2),
    ("2023-01-01 10:10:00", 3),
    ("2023-01-01 10:15:00", 4),
    ("2023-01-01 10:20:00", 5)
]
df = spark.createDataFrame(data, ["timestamp", "value"])

# Преобразуем строки в метки времени
df = df.withColumn("timestamp", col("timestamp").cast("timestamp"))

# Агрегируем данные по скользящим окнам
result = df.groupBy(
    sliding_window("timestamp", "10 minutes", "5 minutes")
).agg(
    count("value").alias("count")
)
result.show(truncate=False)

# Результат:
# +------------------------------------------+-----+
# |window                                    |count|
# +------------------------------------------+-----+
# |[2023-01-01 10:00:00, 2023-01-01 10:10:00]|2    |
# |[2023-01-01 10:05:00, 2023-01-01 10:15:00]|3    |
# |[2023-01-01 10:10:00, 2023-01-01 10:20:00]|3    |
# |[2023-01-01 10:15:00, 2023-01-01 10:25:00]|2    |
# |[2023-01-01 10:20:00, 2023-01-01 10:30:00]|1    |
# +------------------------------------------+-----+

Примечания

• Окна могут перекрываться
• Размер окна и шаг скольжения фиксированные
• Полезно для:
• Анализа трендов
• Сглаживания данных
• Выявления паттернов
• Связанные функции:
• tumbling_window() - фиксированное окно
• window() - общее окно

XML функции

• from_xml(col, schema)
• schema_of_xml(xml)
• to_xml(col)

aggregate functions

• any_value(col[, ignoreNulls])
• approxCountDistinct(col[, rsd])
• approx_percentile(col, percentage[, accuracy])
• array_agg(col)
• avg(col)
• bit_and(col)
• bit_or(col)
• bit_xor(col)
• bool_and(col)
• bool_or(col)
• bool_xor(col)
• collect_list(col)
• collect_set(col)
• corr(col1, col2)
• count(col)
• countDistinct(col, *cols)
• count_distinct(col, *cols)
• count_if(col)
• count_min_sketch(col, eps, confidence, seed)
• covar_pop(col1, col2)
• covar_samp(col1, col2)
• every(col)
• first(col[, ignorenulls])
• first_value(col[, ignoreNulls])
• grouping(col)
• grouping_id(*cols)
• histogram_numeric(col, nBins)
• hll_sketch_agg(col[, lgConfigK])
• hll_union_agg(col[, allowDifferentLgConfigK])
• kurtosis(col)
• last(col[, ignorenulls])
• last_value(col[, ignoreNulls])
• max(col)
• max_by(col, ord)
• mean(col)
• median(col)
• min(col)
• min_by(col, ord)
• mode(col)
• percentile(col, percentage)
• percentile_approx(col, percentage, accuracy=10000)
• product(col)
• regr_avgx(y, x)
• regr_avgy(y, x)
• regr_count()
• regr_intercept(y, x)
• regr_r2()
• regr_slope(y, x)
• regr_syy()
• skewness(col)
• some(col)
• stddev(col)
• stddev_pop(col)
• stddev_samp(col)
• sum(col)
• sum_distinct(col)
• var_pop(col)
• var_samp(col)
• variance(col)

array functions pyspark

• array(*cols)
• array_append(col, value)
• array_compact(col)
• array_contains(col, value)
• array_distinct(col)
• array_insert(arr, pos, value)
• array_intersect(col1, col2)
• array_join(col, delimiter[, null_replacement])
• array_max(col)
• array_min(col)
• array_position(col, value)
• array_prepend(col, value)
• array_remove(col, element)
• array_repeat(col, count)
• array_size(col)
• array_sort(col[, comparator])
• array_union(col1, col2)
• arrays_overlap(a1, a2)
• arrays_zip(*cols)
• cardinality(column)
• concat(*cols)
• element_at(col, extraction)
• exists(col, f)
• explode(column)
• explode_outer(column)
• flatten(col)
• forall(col, f)
• inline(column)
• inline_outer(column)
• json_array_length(json_array)
• json_object_keys(json_object)
• map_filter(map, f)
• map_zip_with(map1, map2, function)
• posexplode(column)
• posexplode_outer(column)
• reduce(col, initialValue, merge, finish=None)
• reverse(col)
• sequence(start, stop[, step])
• shuffle(col)
• size(col)
• slice(x, start, length)
• sort_array(column, asc=True)
• stack(n, expr1, ..., exprN)
• struct(*cols)
• transform(col, f)
• transform_keys(map, f)
• transform_values(map, f)
• try_element_at()
• zip_with(left, right, f)

basic functions

• col(colName)
• column(colName)
• lit(value)

bitmap functions

• bitmap_construct_agg(col)
• bitmap_count(bitmap)
• bitmap_or_agg(bitmap)
• bitwise_not(col)
• bitwise_or(col1, col2)
• bitwise_xor(col1, col2)

bitwise functions

• bit_count()
• bit_get()
• bit_set(column, pos)
• bitwiseAND(column1, column2)
• bitwiseNOT(column)
• bitwiseOR(column1, column2)
• bitwiseXOR(column1, column2)
• getbit()
• setbit(column, pos)
• shiftleft(column, numBits)
• shiftright(column, numBits)
• shiftrightunsigned(column, numBits)

call functions

• call_function()
• call_udf()
• pandas_udf()
• udf()
• udtf()
• unwrap_udt()

comparison functions

• least(expr1, expr2, ...)

conditional functions

• when(condition, value)

conversion functions

• to_binary(expr, format)
• to_char(expr, format)
• to_number(expr, format)

csv functions

• from_csv(col, schema[, options])
• schema_of_csv(csv[, options])
• to_csv(col[, options])

datetime functions pyspark

• add_months(start_date, num_months)
• curdate()
• current_date()
• current_time()
• current_timezone()
• date_add()
• date_diff(endDate, startDate)
• date_part(field, source)
• date_sub()
• date_trunc()
• dateadd(unit, value, date)
• datediff(end_date, start_date)
• datepart(field, source)
• day(date)
• dayofmonth(col)
• dayofweek(date)
• dayofyear(date)
• extract(field FROM source)
• from_utc_timestamp(timestamp, timezone)
• hour(timestamp)
• last_day(date)
• localtimestamp()
• make_date(year, month, day)
• make_dt_interval(days, hours, minutes, seconds)
• make_timestamp(year, month, day, hour, min, sec)
• make_timestamp_ltz(year, month, day, hour, minute, second)
• make_ym_interval(years, months)
• minute(timestamp)
• month(date)
• months_between(end_date, start_date)
• next_day(date, day_of_week)
• now()
• quarter(date)
• second(timestamp)
• session_window(timeColumn, gapDuration)
• timestamp_micros(microseconds)
• timestamp_millis(milliseconds)
• timestamp_seconds(seconds)
• timestampadd(unit, interval, timestamp)
• to_date(col, format=None)
• to_timestamp(col, format=None)
• to_timestamp_ltz(col, format=None)
• to_timestamp_ntz(col, format=None)
• to_unix_timestamp(col, format=None)
• to_utc_timestamp(timestamp, timezone)
• trunc(date, format)
• try_to_timestamp(col, format=None)
• unix_date(col)
• unix_micros(col)
• unix_millis(col)
• unix_seconds(col)
• unix_timestamp(timestamp=None, format=None)
• weekday(date)
• weekofyear(date)
• window_time(col)
• year(date)
• years_between(end, start)

encryption functions

• try_aes_decrypt(input, key, mode, padding)

expression functions

• expr(expr)

hash functions

• crc32(col)
• hash(*cols)
• md5(col)
• sha(col)
• sha1(col)
• sha2(col, numBits)
• xxhash64(*cols)

java functions

• java_method(class_name, method_name, *args)
• reflect(class_name, method_name, *args)

json functions

• from_json(col, schema, options={})
• get_json_object()
• json_tuple(col, *fields)
• to_json(col, options={})

map functions

• create_map(*cols)
• map_concat(map1, map2, ...)
• map_contains_key(col, value)
• map_entries(map)
• map_from_arrays(keys, values)
• map_from_entries(array)
• map_keys(map)
• map_values(map)
• str_to_map(text[, pairDelim, keyValueDelim])

mathematical functions pyspark

• abs(col)
• asin(col)
• atan(col)
• atan2(y, x)
• bin(col)
• bround(col, scale=0)
• cbrt(col)
• ceil(col)
• ceiling(col)
• conv(num, from_base, to_base)
• cos(col)
• cosh(col)
• cot(col)
• csc(col)
• e()
• exp()
• expm1()
• factorial()
• floor(col)
• hypot(col1, col2)
• log(col)
• log10(col)
• negative(col)
• pmod(dividend, divisor)
• positive(col)
• power(col1, col2)
• rand(seed=None)
• randn(seed=None)
• rint(col)
• round(col, scale=0)
• sec(col)
• sign(col)
• signum(col)
• tan(col)
• tanh(col)
• toDegrees(col)
• try_add()
• try_avg()
• try_divide()
• try_multiply()
• try_subtract()
• try_sum()
• try_to_number()

normal functions

• greatest(*cols)

null functions

• coalesce(expr1, expr2, ...)
• equal_null(expr1, expr2)
• ifnull(expr1, expr2)
• isnotnull(expr)
• isnull(expr)
• nanvl(col1, col2)
• nullif(expr1, expr2)
• nvl(expr1, expr2)
• nvl2(expr1, expr2, expr3)

sketch functions

• hll_sketch_estimate(sketch)
• hll_union(sketch1, sketch2)

sort functions

• asc(col)
• asc_nulls_first(col)
• asc_nulls_last(col)
• desc(col)
• desc_nulls_first(col)
• desc_nulls_last(col)

string functions pyspark

• char_length(col)
• eval(expr)
• left(col, len)
• levenshtein(col1, col2)
• mask(col, upperChar, lowerChar, digitChar, otherChar)
• regexp(col, pattern)
• regexp_extract_all(col, pattern)
• regexp_instr(col, pattern, pos, occurrence)
• regexp_like(col, pattern)
• regexp_substr(col, pattern, pos, occurrence)
• repeat(col, n)
• replace(col, search, replace)
• right(col, len)
• rlike(col, pattern)
• rpad(col, len, pad)
• rtrim(col)
• sentences(str, lang, country)
• soundex(col)
• split(str, pattern)
• split_part(str, delimiter, partNum)
• sql(query)
• startswith(col, prefix)
• substr(col, pos, len)
• substring_index(col, delim, count)
• to_varchar(col)
• translate(col, matchingString, replaceString)
• trim(col)
• ucase(col)
• unbase64(col)
• unhex(col)
• upper(col)
• url_decode(str)
• url_encode(str)
• uuid()

struct functions

• named_struct(name1, val1, name2, val2, ...)

system functions

• current_catalog()
• current_database()
• current_schema()
• current_user()
• input_file_block_length()
• input_file_block_start()
• input_file_name()
• spark_partition_id()
• user()

type functions

• cast()
• typeof()

window functions

• cume_dist()
• dense_rank()
• lag(col, offset, default)
• lead(col, offset, default)
• monotonically_increasing_id()
• nth_value(col, n)
• ntile(n)
• percent_rank()
• rank()
• row_number()
• window(timeColumn, windowDuration, slideDuration, startTime)

xml functions

• xml_tuple(xml, *paths)
• xpath(xml, path)
• xpath_boolean(xml, path)
• xpath_double(xml, path)
• xpath_float(xml, path)
• xpath_int(xml, path)
• xpath_long(xml, path)
• xpath_number(xml, path)
• xpath_short(xml, path)
• xpath_string(xml, path)

агрегатные функции

• approx_count_distinct(col, rsd)
• std(col)
• sumDistinct(col)

битовые операции

• bitwise_and(col1, col2)

оконные функции

• diff(col)
• session_window(timeColumn, gapDuration)
• tumbling_window(timeColumn, windowDuration)

операторы сравнения

• <>(left, right)
• <=>(left, right)
• =(left, right)

предикатные функции

• isnotnan(col)

системные функции

• database()
• session_user()

условные функции

• case_when(condition1, value1, condition2, value2, ..., else_value)
• if(condition, value1, value2)

функции для работы с битовыми картами

• bitmap_bit_position(bitmap, bit)
• bitmap_bucket_number(bitmap)

функции преобразования типов

• to_decimal(col, scale=0)

функции шифрования

• aes_decrypt(col, key)