Forecasting with HEAVY.AI and Prophet

1
conn = ibis.heavyai.connect(
2
    host='metis.mapd.com', user='admin', password='HyperInteractive',
3
    port=443, database='heavyai', protocol= 'https'
4
)
5
6
t = conn.table('flights_donotmodify')

1
def get_delay_ts(city: str):
2
    ds=t.dep_timestamp.truncate('D').name('ds')
3
    y = t.depdelay.mean().name('y')
4
    filters = [t.dest_city == city, 
5
               ds.notnull()]
6
7
    expr = t.filter(filters).group_by(ds).aggregate(y)
8
    return expr

1
print(get_delay_ts('New York').compile())
2
3
#this compiles to
4
SELECT DATE_TRUNC(DAY, "dep_timestamp") AS ds, avg("depdelay") AS y
5
FROM flights_2008_7M
6
WHERE ("dest_city" = 'New York') AND
7
      (DATE_TRUNC(DAY, "dep_timestamp") IS NOT NULL) AND
8
      ("flight_year" >= 2008)
9
GROUP BY ds

1
from fbprophet import Prophet
2
3
m = Prophet()
4
m.add_seasonality(name='yearly', period=365, fourier_order=20)
5
6
m.fit(get_delay_ts('Chicago').execute())
7
future = m.make_future_dataframe(periods=30)
8
forecast = m.predict(future)
9
forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail()

1
connection = ibis.pandas.connect({'forecast': forecast })
2
table = connection.table('forecast')
3
4
line = alt.Chart(table).mark_line().encode(
5
    x='ds:T',
6
    y='yhat:Q'
7
)
8
9
band = alt.Chart(table).mark_errorband().encode(
10
    x='ds:T',
11
    y='yhat_lower:Q',
12
    y2='yhat_upper:Q'
13
)
14
15
points = alt.Chart(get_delay_ts('New York')).mark_point(size=2, color='red').encode(
16
x='ds:T',
17
y='y:Q')
18
19
(band+line+points).properties(width=1000, height=400)

1
t.dest_city.nunique().execute()
2
3
#this should return the following, for the public flights dataset used above
4
290

1
conda install -c conda-forge dask

1
def get_delay_ts(city: str = None):
2
    ds=t.dep_timestamp.truncate('D').name('ds')
3
    y = t.depdelay.mean().name('y')
4
5
    filters = [ds.notnull()]
6
7
    if city is not None:
8
        filters.append(t.dest_city == city)
9
10
    expr = t.filter(filters).group_by([ds, t.dest_city]).aggregate(y)
11
    return expr

1
    print(get_delay_ts().compile())
2
    
3
    #This will return the following. Passing in a city parameter will filter by the city
4
    SELECT DATE_TRUNC(DAY, "dep_timestamp") AS ds, "dest_city",
5
       avg("depdelay") AS y
6
       FROM flights_donotmodify
7
       WHERE DATE_TRUNC(DAY, "dep_timestamp") IS NOT NULL
8
    GROUP BY ds, dest_city

1
from dask import delayed
2
from dask.distributed import Client, progress
3
4
def get_forecast_for_ts(city, ts):
5
    print(f'Computing forecast for {city}')
6
    m = Prophet()
7
    m.add_seasonality(name='yearly', period=365, fourier_order=10)
8
    m.fit(ts)
9
    future = m.make_future_dataframe(periods=7, freq='D')
10
    fcst = m.predict(future)
11
    return fcst
12
    
13
#This function creates an input dataframe for Prophet, keyed by city
14
#and executes the prophet forecast in parallel over a specified number of workers
15
def get_forecasts_dask(city:str = None):
16
    input_df = get_delay_ts(city=city).execute()
17
    
18
    #Change n_workers as needed for the machine you're running on (number of cores)
19
    client = Client(threads_per_worker=1, n_workers=16)
20
21
    #Use dask delayed, to create an array of deferred calls to the forecast method, splitting the above result dataframe by city
22
    fcsts = [delayed(get_forecast_for_ts)(city=k, ts=v[['ds','y']]) for k, v in input_df.groupby('dest_city')]
23
    
24
    #Run the forecasts in parallel
25
    try:
26
        forecasts = delayed(fcsts).compute()
27
        return forecasts
28
    except Exception as e:
29
        print(f"Unexpected error {e}")
30
    finally:
31
        client.close()

1
%%time
2
dask_forecasts=get_forecasts_dask()

1
CPU times: user 27 s, sys: 1.91 s, total: 28.9 s
2
Wall time: 1min 31s

1
pip install -U ray

1
import ray
2
3
#the below decorator allows Ray's runtime to execute multiple instances of it in parallel
4
@ray.remote
5
def get_forecast_for_ts(city, ts):
6
    print(f'Computing forecast for {city}')
7
    m = Prophet()
8
    m.add_seasonality(name='yearly', period=365, fourier_order=10)
9
    m.fit(ts)
10
    future = m.make_future_dataframe(periods=7, freq='D')
11
    fcst = m.predict(future)
12
    return fcst
13
14
#This method is very similar to the Dask method, except for how Ray is invoked
15
def get_forecasts_ray(city:str = None):
16
    ray.shutdown()
17
    ray.init(ignore_reinit_error=True, log_to_driver=False, logging_level='INFO')
18
    
19
    #Use Ibis to get the specific set of fuel locations to run the forecast on
20
    input_df = get_delay_ts(city=city).execute()
21
22
    #Create an array of deferred forecast objects that Ray will evaluate in parallel
23
    fcsts = [get_forecast_for_ts.remote(city=k, ts=v[['ds','y']]) for k, v in input_df.groupby('dest_city')]
24
    return ray.get(fcsts)

1
CPU times: user 16.4 s, sys: 762 ms, total: 17.1 s
2
Wall time: 1min 23s

1
2020-08-30 16:20:02,762	INFO resource_spec.py:231 -- Starting Ray with 29.54 GiB memory available for workers and up to 14.77 GiB for objects. You can adjust these settings with ray.init(memory=<bytes>, object_store_memory=<bytes>).
2
2020-08-30 16:20:03,142	INFO services.py:1193 -- View the Ray dashboard at localhost:8265
3
4
CPU times: user 16.4 s, sys: 762 ms, total: 17.1 s
5
Wall time: 1min 23s