通过

series_fbprophet_forecast_fl()

  • 项目

适用于：✅Azure 数据资源管理器

函数 series_fbprophet_forecast_fl() 是一个用户定义的函数 (UDF)，它使用包含时序的表达式作为输入，并使用 Prophet 算法预测最后一个尾随点的值。 该函数将返回预测点及其置信区间。 此函数是 Prophet() 类的 Kusto 查询语言 (KQL) 包装器，只公开预测所必需的参数。 可随意修改副本以支持更多参数。 例如假期、更改点、傅立叶顺序等。

注意

请考虑使用原生函数 series_decompose_forecast()。 该原生函数基于一个更简单的模型，但其可伸缩性更强且运行速度更快。

先决条件

  • 必须在群集上启用 Python 插件。 这是函数中使用的内联 Python 所必需的。
  • 安装 fbprophet 包，因为它未包括在 Python 映像中。 若要安装该包，请执行以下操作：
    1. 请遵循安装 Python 插件的包中的指南。
    2. 创建对 Zip 文件具有读取访问权限的 SAS 令牌。 若要创建 SAS 令牌，请参阅获取 Blob 容器的 SAS。
    3. 在示例中，请将 external_artifacts 参数中的 URL 引用替换为你自己的文件路径及其 SAS 令牌。

语法

T | invoke series_fbprophet_forecast_fl(ts_series, y_series, y_pred_series, [ points ], [ y_pred_low_series ], [ y_pred_high_series ])

详细了解语法约定。

参数

客户 类型​​ 必需 说明
ts_series string ✔️ 输入表列的名称，其中包含要预测的序列的时间戳。
y_series string ✔️ 输入表列的名称，其中包含要预测的序列的值。
y_pred_series string ✔️ 存储已预测序列的列的名称。
points int ✔️ 要预测的序列末端的点数。 这些点将从学习（回归）过程中排除。 默认值为 0。
y_pred_low_series string 用于存储置信区间最小值序列的列的名称。 如果不需要置信区间，则省略此项。
y_pred_high_series string 用于存储置信区间最大值序列的列的名称。 如果不需要置信区间，则省略此项。

函数定义

可以通过将函数的代码嵌入为查询定义的函数，或将其创建为数据库中的存储函数来定义函数，如下所示：

使用以下 let 语句定义函数。 不需要任何权限。

重要

let 语句不能独立运行。 它必须后跟一个表格表达式语句。 若要运行 series_fbprophet_forecast_fl() 的工作示例，请参阅示例。

let series_fbprophet_forecast_fl=(tbl:(*), ts_series:string, y_series:string, y_pred_series:string, points:int=0, y_pred_low_series:string='', y_pred_high_series:string='')
{
    let kwargs = bag_pack('ts_series', ts_series, 'y_series', y_series, 'y_pred_series', y_pred_series, 'points', points, 'y_pred_low_series', y_pred_low_series, 'y_pred_high_series', y_pred_high_series);
    let code = ```if 1:
        from sandbox_utils import Zipackage
        Zipackage.install("prophet.zip")
        ts_series = kargs["ts_series"]
        y_series = kargs["y_series"]
        y_pred_series = kargs["y_pred_series"]
        points = kargs["points"]
        y_pred_low_series = kargs["y_pred_low_series"]
        y_pred_high_series = kargs["y_pred_high_series"]
        result = df
        sr = pd.Series(df[y_pred_series])
        if y_pred_low_series != '':
            srl = pd.Series(df[y_pred_low_series])
        if y_pred_high_series != '':
            srh = pd.Series(df[y_pred_high_series])
        from prophet import Prophet
        df1 = pd.DataFrame(columns=["ds", "y"])
        for i in range(df.shape[0]):
            df1["ds"] = pd.to_datetime(df[ts_series][i])
            df1["ds"] = df1["ds"].dt.tz_convert(None)
            df1["y"] = df[y_series][i]
            df2 = df1[:-points]
            m = Prophet()
            m.fit(df2)
            future = df1[["ds"]]
            forecast = m.predict(future)
            sr[i] = list(forecast["yhat"])
            if y_pred_low_series != '':
                srl[i] = list(forecast["yhat_lower"])
            if y_pred_high_series != '':
                srh[i] = list(forecast["yhat_upper"])
        result[y_pred_series] = sr
        if y_pred_low_series != '':
            result[y_pred_low_series] = srl
        if y_pred_high_series != '':
            result[y_pred_high_series] = srh
    ```;
    tbl
     | evaluate python(typeof(*), code, kwargs
, external_artifacts=bag_pack('prophet.zip', 'https://artifcatswestus.blob.core.windows.net/public/prophet-1.1.5.zip?*** YOUR SAS TOKEN ***'))
};
// Write your query to use the function here.

示例

以下示例使用 invoke 运算符运行函数。

若要使用查询定义的函数，请在嵌入的函数定义后调用它。

let series_fbprophet_forecast_fl=(tbl:(*), ts_series:string, y_series:string, y_pred_series:string, points:int=0, y_pred_low_series:string='', y_pred_high_series:string='')
{
    let kwargs = bag_pack('ts_series', ts_series, 'y_series', y_series, 'y_pred_series', y_pred_series, 'points', points, 'y_pred_low_series', y_pred_low_series, 'y_pred_high_series', y_pred_high_series);
    let code = ```if 1:
        from sandbox_utils import Zipackage
        Zipackage.install("prophet.zip")
        ts_series = kargs["ts_series"]
        y_series = kargs["y_series"]
        y_pred_series = kargs["y_pred_series"]
        points = kargs["points"]
        y_pred_low_series = kargs["y_pred_low_series"]
        y_pred_high_series = kargs["y_pred_high_series"]
        result = df
        sr = pd.Series(df[y_pred_series])
        if y_pred_low_series != '':
            srl = pd.Series(df[y_pred_low_series])
        if y_pred_high_series != '':
            srh = pd.Series(df[y_pred_high_series])
        from prophet import Prophet
        df1 = pd.DataFrame(columns=["ds", "y"])
        for i in range(df.shape[0]):
            df1["ds"] = pd.to_datetime(df[ts_series][i])
            df1["ds"] = df1["ds"].dt.tz_convert(None)
            df1["y"] = df[y_series][i]
            df2 = df1[:-points]
            m = Prophet()
            m.fit(df2)
            future = df1[["ds"]]
            forecast = m.predict(future)
            sr[i] = list(forecast["yhat"])
            if y_pred_low_series != '':
                srl[i] = list(forecast["yhat_lower"])
            if y_pred_high_series != '':
                srh[i] = list(forecast["yhat_upper"])
        result[y_pred_series] = sr
        if y_pred_low_series != '':
            result[y_pred_low_series] = srl
        if y_pred_high_series != '':
            result[y_pred_high_series] = srh
    ```;
    tbl
     | evaluate python(typeof(*), code, kwargs
, external_artifacts=bag_pack('prophet.zip', 'https://artifactswestus.blob.core.windows.net/public/prophet-1.1.5.zip?*** YOUR SAS TOKEN ***'))
};
//
//  Forecasting 3 time series using fbprophet, compare to forecasting using the native function series_decompose_forecast()
//
let min_t = datetime(2017-01-05);
let max_t = datetime(2017-02-03 22:00);
let dt = 2h;
let horizon=7d;
demo_make_series2
| make-series num=avg(num) on TimeStamp from min_t to max_t+horizon step dt by sid 
| extend pred_num_native = series_decompose_forecast(num, toint(horizon/dt))
| extend pred_num=dynamic(null), pred_num_lower=dynamic(null), pred_num_upper=dynamic(null)
| invoke series_fbprophet_forecast_fl('TimeStamp', 'num', 'pred_num', toint(horizon/dt), 'pred_num_lower', 'pred_num_upper')
| render timechart

输出

显示预测几个时序的关系图。