Scalable Time Series Forecasting using Apache Spark and Facebook Prophet
Time series forecasting is a critical task in many domains, from retail demand planning to sensor anomaly detection to financial market prediction. The goal is to predict future values of a quantity based on its historical values, accounting for trends, seasonality, and other patterns in the data.
As time series datasets grow to terabytes or petabytes in size, traditional forecasting methods struggle to keep up. Fortunately, modern big data tools like Apache Spark provide a powerful platform for scalable time series modeling. When combined with a high-quality forecasting library like Facebook Prophet, Spark enables building thousands of time series models in parallel across a cluster of machines.
In this post, we‘ll walk through a complete example of using Spark and Prophet together to forecast product demand for a large retailer. We‘ll cover:
- Loading and preprocessing a multi-entity time series dataset using PySpark
- Partitioning the data by entity and building models in parallel using Spark
- Applying the Prophet library to each partition using Spark UDFs
- Visualizing, interpreting and evaluating the results
- Tuning Spark performance for time series model building
By the end, you‘ll see how Spark and Prophet provide a scalable, flexible and performant platform for enterprise time series use cases. Let‘s get started!
About Apache Spark
Apache Spark is an open-source unified analytics engine for large-scale data processing. Developed originally at UC Berkeley in 2009, Spark has become the most active open source project in data processing and analytics, with over 1000 contributors from more than 250 organizations.
Spark‘s core strengths include:
- Scalable in-memory computing: Spark can cache datasets in memory and query them repeatedly, providing performance up to 100x faster than Hadoop MapReduce.
- General purpose batch and streaming: Spark provides batch processing, real-time streaming, machine learning, and graph processing APIs, allowing diverse workloads to be combined.
- Multilanguage support: Spark provides APIs in Java, Scala, Python, R, and SQL, making it accessible to a wide audience.
- Integration with big data tools: Spark can read from and write to a variety of data sources and processing engines in the Hadoop ecosystem.
For time series forecasting, Spark‘s key benefits are its ability to manipulate large datasets using SQL-like APIs, cache datasets in memory for fast iterative queries, and scale processing across clusters of machines. This makes it an ideal processing engine for applying advanced time series models to big data.
About Facebook Prophet
Prophet is an open-source library for time series forecasting developed by Facebook‘s Data Science team. It provides an intuitive, human-friendly approach to forecasting based on an additive model where non-linear trends are fit with seasonality and holiday effects.
Prophet‘s key features include:
- Automatic seasonality detection: Prophet determines seasonality from the data, allowing it to capture weekly and yearly patterns without manual configuration.
- Robust handling of missing data and outliers: Prophet is robust to outliers and handles missing data well, making it applicable to messy real-world data.
- Interpretable parameters: Prophet uses clear parameters like "changepoints" that are easy for analysts to understand and set.
- Extensibility: Prophet allows analysts to bring in their own knowledge by manually specifying seasonality and holiday effects.
While originally implemented in R, Prophet now has an official Python API that integrates easily with tools like Spark. Overall, it provides an advanced but intuitive approach for forecasting that lends itself well to enterprise settings.
Dataset and Setup
For this example, we‘ll use a synthetic demand forecasting dataset that represents 500 weeks of sales data for multiple stores of a large retailer. The raw data looks like:
date,store_id,sales 2013-01-01,1,4924 2013-01-08,1,4722 ...
To work with this in Spark, we first need to load the data into a PySpark DataFrame. Assuming Spark is already configured, we can create a SparkSession and read the CSV data:
from pyspark.sql import SparkSessionspark = SparkSession.builder \ .appName("TimeSeries") \ .getOrCreate()
df = spark.read.csv("sales_data.csv", header=True, inferSchema=True)
This creates a DataFrame with columns for date, store ID, and sales value. We can inspect the schema and first few rows:
df.printSchema() df.show(5)
Before building models, it‘s often helpful to visualize the raw time series data. We can use the ‘.toPandas()‘ method to collect the data into a Pandas DataFrame for easy plotting:
import matplotlib.pyplot as pltpdf = df.toPandas() pdf.set_index(‘date‘, inplace=True)
fig, ax = plt.subplots(1, 1, figsize=(15,5))
for store_id in pdf.store_id.unique(): mask = pdf.store_id == store_id pdf[mask][[‘sales‘]].plot(ax=ax, label=f‘Store {store_id}‘)
ax.set_xlabel(‘Date‘)
ax.set_ylabel(‘Sales‘) ax.legend()plt.show()
This plots the sales data for each store ID on a shared axis, allowing us to see both overall trends and store-specific patterns. We observe some global seasonality, but each store also has its own patterns.
Parallel Time Series Modeling
To build time series models per store, we need to partition the DataFrame by the ‘store_id‘ column. In Spark, this can be done using the ‘.repartition‘ method:
store_df = df.repartition(‘store_id‘)
This shuffles the data across workers so that all rows for each store are on the same partition. We can then apply a modeling function to each store partition in parallel using Spark‘s User-Defined Function (UDF) functionality.
First, we define a function to apply a Prophet model to each group of rows in the repartitioned DataFrame:
from fbprophet import Prophet from pyspark.sql.functions import pandas_udf from pyspark.sql.types import *result_schema = StructType([ StructField(‘ds‘,TimestampType()), StructField(‘store_id‘,IntegerType()), StructField(‘y‘,FloatType()), StructField(‘yhat‘,FloatType()), StructField(‘yhat_upper‘,FloatType()), StructField(‘yhat_lower‘,FloatType()) ])
@pandas_udf(result_schema, PandasUDFType.GROUPED_MAP)
def forecast_sales(keys, df):# instantiate the model model = Prophet( changepoint_prior_scale=0.05, seasonality_prior_scale=10, holidays_prior_scale=10, seasonality_mode=‘multiplicative‘, weekly_seasonality=True, yearly_seasonality=True ) # rename columns to match Prophet schema df = df.rename(columns={ ‘date‘:‘ds‘, ‘sales‘:‘y‘ }) # fit the model model.fit(df) # generate predictions future_pd = model.make_future_dataframe(periods=12, freq=‘W‘) forecast_pd = model.predict(future_pd) # merge predictions with historical data results_pd = forecast_pd[[‘ds‘,‘yhat‘,‘yhat_upper‘,‘yhat_lower‘]].set_index(‘ds‘) results_pd = results_pd.join(df[[‘ds‘,‘store_id‘,‘y‘]].set_index(‘ds‘), how=‘left‘) results_pd[‘store_id‘] = keys[0] return results_pd[[‘ds‘,‘store_id‘,‘y‘,‘yhat‘,‘yhat_upper‘,‘yhat_lower‘]]This function takes a store_id key and a DataFrame of sales data for that store. It renames the columns to match Prophet‘s expected schema, fits the model to the historical data, and generates 12 weeks of forecast values. Finally, it merges the forecasts with the historical data and returns the full results.
The ‘@pandas_udf‘ decorator specifies that this function will operate on DataFrames (rather than individual rows) for each store_id key. We also provide a result schema that matches the output columns of the function.
With the UDF defined, we can now apply it to the partitioned DataFrame:
results = ( store_df .groupby(‘store_id‘) .apply(forecast_sales) )This fits a separate model for each store partition and returns a new DataFrame with the store-level sales forecasts. We can view the results with:
results.select(‘store_id‘,‘ds‘,‘y‘,‘yhat‘).show(20)To visualize the forecasts, we again collect to a Pandas DataFrame and plot:
results_pd = results.toPandas() results_pd[‘ds‘] = pd.to_datetime(results_pd[‘ds‘])fig, axs = plt.subplots(2, 2, figsize=(15, 10))
for i, store_id in enumerate(results_pd.store_id.unique()): ax = axs[i//2, i%2]
mask = results_pd.store_id == store_id results_pd[mask].plot(x=‘ds‘, y=[‘y‘, ‘yhat‘], ax=ax) ax.fill_between( results_pd[mask].ds, results_pd[mask].yhat_lower, results_pd[mask].yhat_upper, alpha=0.2, color=‘gray‘ ) ax.set_title(f‘Store {store_id}‘) ax.set_xlabel(‘Date‘) ax.set_ylabel(‘Sales‘) ax.legend()plt.tight_layout()
plt.show()This plots the actual and predicted sales for a sample of stores, along with Prophet‘s uncertainty interval. We see that the forecasts align well with the actual sales data, capturing key trends and seasonality for each store. The model seems to perform better for some stores than others, so further tuning may improve the results.
Performance Tuning
There are a number of ways to optimize the performance of this Spark forecasting job:
- Persist and cache: Persist and cache the partitioned DataFrame in memory before applying the model function to avoid recomputing the partitioning.
- Increase parallelism: Increase the number of partitions in the DataFrame to process more stores concurrently. The optimal number of partitions depends on the size of the cluster and the data.
- Use vectorized UDFs: Spark UDFs are a frequent bottleneck, so using vectorized Pandas UDFs (as we did) rather than row-at-a-time UDFs can dramatically improve performance.
- Tune the resource configuration: Adjust the memory and CPU resource allocation for the Spark driver and executor pods. In particular, using more memory can help avoid expensive disk spills.
Conclusion
This post demonstrated how to use Apache Spark and Facebook Prophet to build time series forecasting models at scale. We covered:
- Loading and partitioning a multi-entity time series dataset using PySpark
- Defining a vectorized UDF to apply Prophet models to each partition
- Visualizing and interpreting the model results
- Tuning Spark performance for better model building speed
Using Spark and Prophet together, data scientists can build thousands of granular time series models for large datasets in a scalable, distributed fashion. This pattern could be applied to a wide variety of enterprise forecasting use cases, from retail demand planning to server metrics prediction to economic forecasting.
Notably, while Prophet is an advanced library, there are many alternative time series packages that integrate with Spark. For example, spark-ts provides native Scala and Java APIs for common time series methods like ARIMA, Holt-Winters Exponential Smoothing, and Generalized Linear Models. The principles covered here extend to using any of these libraries with Spark.
I hope this post was helpful for getting started applying distributed time series models to large datasets! Feel free to connect with me on LinkedIn with any questions or thoughts. You can find the complete code for this example on Github. Thanks for reading!
