tempo.tsdf#

Note

Some of the function docstrings are not in line with the standard RST pattern so they maybe renderred incorrectly. We are fixing them. They will be updated very soon. Thank you for your patience and support.

class tempo.tsdf.TSDF(df: DataFrame, ts_col: str = 'event_ts', partition_cols: Optional[list[str]] = None, sequence_col: Optional[str] = None)[source]#

This object is the main wrapper over a Spark data frame which allows a user to parallelize time series computations on a Spark data frame by various dimensions. The two dimensions required are partition_cols (list of columns by which to summarize) and ts_col (timestamp column, which can be epoch or TimestampType).

EMA(colName, window=30, exp_factor=0.2)[source]#

Constructs an approximate EMA in the fashion of: EMA = e * lag(col,0) + e * (1 - e) * lag(col, 1) + e * (1 - e)^2 * lag(col, 2) etc, up until window TODO: replace case when statement with coalesce TODO: add in time partitions functionality (what is the overlap fraction?)

after(ts)[source]#

Select only records after a given time

Parameters

ts – timestamp on which to filter records

Returns

a TSDF object containing just the records after the given time

asofJoin(right_tsdf, left_prefix=None, right_prefix='right', tsPartitionVal=None, fraction=0.5, skipNulls=True, sql_join_opt=False, suppress_null_warning=False)[source]#

Performs an as-of join between two time-series. If a tsPartitionVal is specified, it will do this partitioned by time brackets, which can help alleviate skew.

NOTE: partition cols have to be the same for both Dataframes. We are collecting stats when the WARNING level is enabled also.

Parameters :param right_tsdf - right-hand data frame containing columns to merge in :param left_prefix - optional prefix for base data frame :param right_prefix - optional prefix for right-hand data frame :param tsPartitionVal - value to break up each partition into time brackets :param fraction - overlap fraction :param skipNulls - whether to skip nulls when joining in values :param sql_join_opt - if set to True, will use standard Spark SQL join if it is estimated to be efficient :param suppress_null_warning - when tsPartitionVal is specified, will collect min of each column and raise warnings about null values, set to True to avoid

at(ts)[source]#

Select only records at a given time

Parameters

ts – timestamp of the records to select

Returns

a TSDF object containing just the records at the given time

atOrAfter(ts)[source]#

Select only records at or after a given time

Parameters

ts – timestamp on which to filter records

Returns

a TSDF object containing just the records at or after the given time

atOrBefore(ts)[source]#

Select only records at or before a given time

Parameters

ts – timestamp on which to filter records

Returns

a TSDF object containing just the records at or before the given time

before(ts)[source]#

Select only records before a given time

Parameters

ts – timestamp on which to filter records

Returns

a TSDF object containing just the records before the given time

between(start_ts, end_ts, inclusive=True)[source]#

Select only records in a given range

Parameters
  • start_ts – starting time of the range to select

  • end_ts – ending time of the range to select

  • inclusive (bool) – whether the range is inclusive of the endpoints or not, defaults to True

Returns

a TSDF object containing just the records within the range specified

describe()[source]#

Describe a TSDF object using a global summary across all time series (anywhere from 10 to millions) as well as the standard Spark data frame stats. Missing vals Summary global - unique time series based on partition columns, min/max times, granularity - lowest precision in the time series timestamp column count / mean / stddev / min / max - standard Spark data frame describe() output missing_vals_pct - percentage (from 0 to 100) of missing values.

earliest(n: int = 1)[source]#

Select the earliest n records for each series

Parameters

n – number of records to select (default is 1)

Returns

a TSDF object containing the earliest n records for each series

extractStateIntervals(*metric_cols: str, state_definition: Union[str, Callable[[pyspark.sql.column.Column, pyspark.sql.column.Column], pyspark.sql.column.Column]] = '=') pyspark.sql.dataframe.DataFrame[source]#

Extracts intervals from a TSDF based on some notion of “state”, as defined by the :param state_definition: parameter. The state definition consists of a comparison operation between the current and previous values of a metric. If the comparison operation evaluates to true across all metric columns, then we consider both points to be in the same “state”. Changes of state occur when the comparison operator returns false for any given metric column. So, the default state definition (‘=’) entails that intervals of time wherein the metrics all remained constant. A state definition of ‘>=’ would extract intervals wherein the metrics were all monotonically increasing.

Param

metric_cols: the set of metric columns to evaluate for state changes

Param

state_definition: the comparison function used to evaluate individual metrics for state changes.

Either a string, giving a standard PySpark column comparison operation, or a binary function with the signature: (x1: Column, x2: Column) -> Column where the returned column expression evaluates to a BooleanType

Returns

a DataFrame object containing the resulting intervals

fourier_transform(timestep, valueCol)[source]#

Function to fourier transform the time series to its frequency domain representation. :param timestep: timestep value to be used for getting the frequency scale :param valueCol: name of the time domain data column which will be transformed

interpolate(freq: str, func: str, method: str, target_cols: Optional[List[str]] = None, ts_col: Optional[str] = None, partition_cols: Optional[List[str]] = None, show_interpolated: bool = False, perform_checks: bool = True)[source]#

Function to interpolate based on frequency, aggregation, and fill similar to pandas. Data will first be aggregated using resample, then missing values will be filled based on the fill calculation.

Parameters
  • freq – frequency for upsample - valid inputs are “hr”, “min”, “sec” corresponding to hour, minute, or second

  • func – function used to aggregate input

  • method – function used to fill missing values e.g. linear, null, zero, bfill, ffill

  • [optional] (show_interpolated) – columns that should be interpolated, by default interpolates all numeric columns

  • [optional] – specify other ts_col, by default this uses the ts_col within the TSDF object

  • [optional] – specify other partition_cols, by default this uses the partition_cols within the TSDF object

  • [optional] – if true will include an additional column to show which rows have been fully interpolated.

  • perform_checks – calculate time horizon and warnings if True (default is True)

Returns

new TSDF object containing interpolated data

latest(n: int = 1)[source]#

Select the latest n records for each series

Parameters

n – number of records to select (default is 1)

Returns

a TSDF object containing the latest n records for each series

priorTo(ts, n: int = 1)[source]#

Select the n most recent records prior to a given time You can think of this like an ‘asOf’ select - it selects the records as of a particular time

Parameters
  • ts – timestamp on which to filter records

  • n – number of records to select (default is 1)

Returns

a TSDF object containing the n records prior to the given time

resample(freq, func=None, metricCols=None, prefix=None, fill=None, perform_checks=True)[source]#

function to upsample based on frequency and aggregate function similar to pandas :param freq: frequency for upsample - valid inputs are “hr”, “min”, “sec” corresponding to hour, minute, or second :param func: function used to aggregate input :param metricCols supply a smaller list of numeric columns if the entire set of numeric columns should not be returned for the resample function :param prefix - supply a prefix for the newly sampled columns :param fill - Boolean - set to True if the desired output should contain filled in gaps (with 0s currently) :param perform_checks: calculate time horizon and warnings if True (default is True) :return: TSDF object with sample data using aggregate function

select(*cols)[source]#

pyspark.sql.DataFrame.select() method’s equivalent for TSDF objects Parameters ———- cols : str or list of strs column names (string). If one of the column names is ‘*’, that column is expanded to include all columns in the current TSDF.

tsdf.select(‘*’).collect() [Row(age=2, name=’Alice’), Row(age=5, name=’Bob’)] tsdf.select(‘name’, ‘age’).collect() [Row(name=’Alice’, age=2), Row(name=’Bob’, age=5)]

show(n=20, k=5, truncate=True, vertical=False)[source]#

pyspark.sql.DataFrame.show() method’s equivalent for TSDF objects

n : int, optional Number of rows to show. truncate : bool or int, optional If set to True, truncate strings longer than 20 chars by default. If set to a number greater than one, truncates long strings to length truncate and align cells right. vertical : bool, optional If set to True, print output rows vertically (one line per column value).

from pyspark.sql.functions import *

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”))

from tempo import *

phone_accel_tsdf = TSDF(phone_accel_df, ts_col=”event_ts”, partition_cols = [“User”])

# Call show method here phone_accel_tsdf.show()

subsequentTo(ts, n: int = 1)[source]#

Select the n records subsequent to a give time

Parameters
  • ts – timestamp on which to filter records

  • n – number of records to select (default is 1)

Returns

a TSDF object containing the n records subsequent to the given time

withGroupedStats(metricCols=[], freq=None)[source]#

Create a wider set of stats based on all numeric columns by default Users can choose which columns they want to summarize also. These stats are: mean/count/min/max/sum/std deviation :param metricCols - list of user-supplied columns to compute stats for. All numeric columns are used if no list is provided :param freq - frequency (provide a string of the form ‘1 min’, ‘30 seconds’ and we interpret the window to use to aggregate

withLookbackFeatures(featureCols, lookbackWindowSize, exactSize=True, featureColName='features')[source]#

Creates a 2-D feature tensor suitable for training an ML model to predict current values from the history of some set of features. This function creates a new column containing, for each observation, a 2-D array of the values of some number of other columns over a trailing “lookback” window from the previous observation up to some maximum number of past observations.

Parameters
  • featureCols – the names of one or more feature columns to be aggregated into the feature column

  • lookbackWindowSize – The size of lookback window (in terms of past observations). Must be an integer >= 1

  • exactSize – If True (the default), then the resulting DataFrame will only include observations where the generated feature column contains arrays of length lookbackWindowSize. This implies that it will truncate observations that occurred less than lookbackWindowSize from the start of the timeseries. If False, no truncation occurs, and the column may contain arrays less than lookbackWindowSize in length.

  • featureColName – The name of the feature column to be generated. Defaults to “features”

Returns

a DataFrame with a feature column named featureColName containing the lookback feature tensor

withPartitionCols(partitionCols: list[str])[source]#

Sets certain columns of the TSDF as partition columns. Partition columns are those that differentiate distinct timeseries from each other. :param partitionCols: a list of columns used to partition distinct timeseries :return: a TSDF object with the given partition columns

withRangeStats(type='range', colsToSummarize=[], rangeBackWindowSecs=1000)[source]#

Create a wider set of stats based on all numeric columns by default Users can choose which columns they want to summarize also. These stats are: mean/count/min/max/sum/std deviation/zscore :param type - this is created in case we want to extend these stats to lookback over a fixed number of rows instead of ranging over column values :param colsToSummarize - list of user-supplied columns to compute stats for. All numeric columns are used if no list is provided :param rangeBackWindowSecs - lookback this many seconds in time to summarize all stats. Note this will look back from the floor of the base event timestamp (as opposed to the exact time since we cast to long) Assumptions:

  1. The features are summarized over a rolling window that ranges back

  2. The range back window can be specified by the user

  3. Sequence numbers are not yet supported for the sort

  4. There is a cast to long from timestamp so microseconds or more likely breaks down - this could be more easily handled with a string timestamp or sorting the timestamp itself. If using a ‘rows preceding’ window, this wouldn’t be a problem