repository·master·Indexed 43 days ago
https://github.com/databrickslabs/tempoA suite of scalable time series manipulation utilities built on PySpark for the Databricks/Spark ecosystem. It introduces the TSDF (Time Series Data Frame) abstraction to simplify operations such as AS OF joins, resampling, interpolation, rolling statistics (EMA and SMA), and Fourier Transforms.
Tempo is a suite of utilities designed for time series manipulation within Apache Spark. It extends PySpark capabilities by providing specialized abstractions and functions that make working with time series data easier and highly scalable for data teams.
Note: The Scala version of Tempo is deprecated and no longer under development; users should use the Python/PySpark implementation.
The primary entry point for all time series analysis in tempo is the TSDF (Time Series Data Frame) object.
A TSDF wraps a Spark DataFrame and represents a collection of many smaller time series, where each individual time series is defined by a unique set of partition keys.
To initialize a TSDF, you must provide:
ts_col): This column is used for sorting fact and source tables.Optional parameters for TSDF creation include:
partition_cols): Columns used to partition the TSDF into granular time series for windowing and sorting.Tempo simplifies time series manipulation in Spark. To begin, you can wrap a standard Spark DataFrame into a TSDF (Time Series DataFrame) by specifying the timestamp column and any partition columns.
Common operations supported by Tempo include:
from pyspark.sql.functions import *
from tempo import *
# Load and prepare your Spark DataFrame
phone_accel_df = spark.read.format("csv") \
.option("header", "true") \
.load("dbfs:/home/tempo/Phones_accelerometer") \
.withColumn("event_ts", (col("Arrival_Time").cast("double")/1000).cast("timestamp")) \
.withColumn("x", col("x").cast("double")) \
.withColumn("y", col("y").cast("double")) \
.withColumn("z", col("z").cast("double")) \
.withColumn("event_ts_dbl", col("event_ts").cast("double"))
# Convert to a Tempo Time Series DataFrame (TSDF)
phone_accel_tsdf = TSDF(phone_accel_df, ts_col="event_ts", partition_cols = ["User"])
display(phone_accel_tsdf)After cloning the repository, it is recommended to use a virtual environment. Install the required modules using pip from the requirements.txt file.
python -m venv venv
source venv/bin/activate # or appropriate activation for your OS
pip install -r requirements.txtA TSDF (Time Series Data Frame) is the core abstraction in Tempo. You can wrap an existing Spark DataFrame into a TSDF by specifying the timestamp column and optional partition columns. Partition columns are used to split the data into more granular time series (e.g., by User or DeviceID) for windowing and sorting operations.
from tempo import *
# phone_accel_df is a standard Spark DataFrame
phone_accel_tsdf = TSDF(phone_accel_df, ts_col="event_ts", partition_cols = ["User"])You can install the Tempo library using pip. This is the recommended method for most users.
$ pip install dbl-tempoTo build the project into a Python wheel distribution, run the setup.py script with the bdist_wheel command from the main project folder.
python setup.py bdist_wheelIf you need the latest source code directly from GitHub, you can install it in editable mode using the following command.
Note: The Scala version of Tempo is deprecated and no longer in development; use the Python version.
$ pip install -e git+https://github.com/databrickslabs/tempo.git#"egg=dbl-tempo&subdirectory=python"Transform a time series to the frequency domain.
Parameters:
timestep: The timestep value used for the frequency scale.valueCol: The name of the time-domain data column to transform.Note: This currently uses a pandas_udf implementation.
ft_df = tsdf.fourier_transform(timestep=1, valueCol="data_col")Use withGroupedStats to compute metrics (min, max, count, mean, stddev, sum) for numeric columns, grouped by partition columns and a specific frequency.
Parameters:
freq: (Required) String representing frequency (e.g., '1 minute', '40 seconds').metricCols: (Optional) List of numeric columns to compute metrics for. If omitted, all numeric columns are used.# Compute stats for column 'y' every 1 minute
grouped_stats = watch_accel_tsdf.withGroupedStats(metricCols=["y"], freq="1 minute")The resample method allows you to change the frequency of your time series.
Parameters:
freq: The target frequency (e.g., '1 minute', '4 hours', '2 days', 'sec', 'min', 'day').func: The aggregation function to use within the time bucket. Supported values: 'floor', 'ceil', 'min', 'max', 'mean'.fill: Set to True to upsample and fill missing values.Aggregation Functions:
floor: Returns the earliest value by timestamp.ceil: Returns the latest value by timestamp.min: Returns the lowest value regardless of timestamp.max: Returns the highest value regardless of timestamp.mean: Returns the average value regardless of timestamp.# Resample to 1-minute intervals using the floor function
resampled_sdf = phone_accel_tsdf.resample(freq='min', func='floor')The interpolate method fills in missing values or NULLs in a series. It can be used independently or chained after a resample operation.
Supported Methods:
zero: Zero fill.null: Null fill.bfill: Backwards fill.ffill: Forwards fill.linear: Linear interpolation.Parameters:
freq: Frequency for resampling before interpolation.func: Aggregation function for resampling ('floor', 'ceil', 'min', 'max', 'mean').target_cols: List of specific columns to interpolate.show_interpolated: If True, adds boolean columns indicating if a row was interpolated.Chaining Pattern:
If chained after resample, you do not need to provide freq or func to interpolate.
# Chaining: Resample to 30s intervals using mean, then linear interpolate
interpolated_tsdf = input_tsdf.resample(freq="30 seconds", func="mean").interpolate(method="linear")
# Independent interpolation with specific columns and metadata
interpolated_tsdf = input_tsdf.interpolate(
freq="30 seconds",
func="mean",
target_cols=["columnA", "columnB"],
method="linear",
show_interpolated=True
)