Java 类名：com.alibaba.alink.operator.stream.timeseries.LSTNetPredictStreamOp
Python 类名：LSTNetPredictStreamOp

功能介绍

使用 LSTNet 进行时间序列训练和预测。

使用方式

参考文档 https://www.yuque.com/pinshu/alink_guide/xbp5ky

参数说明

| 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| predictionCol | 预测结果列名 | 预测结果列名 | String | ✓ | | |

| valueCol | value列，类型为MTable | value列，类型为MTable | String | ✓ | 所选列类型为 [M_TABLE] | |

| modelFilePath | 模型的文件路径 | 模型的文件路径 | String | | | null |

| predictNum | 预测条数 | 预测条数 | Integer | | | 1 |

| predictionDetailCol | 预测详细信息列名 | 预测详细信息列名 | String | | | |

| reservedCols | 算法保留列名 | 算法保留列 | String[] | | | null |

| numThreads | 组件多线程线程个数 | 组件多线程线程个数 | Integer | | | 1 |

| modelStreamFilePath | 模型流的文件路径 | 模型流的文件路径 | String | | | null |

| modelStreamScanInterval | 扫描模型路径的时间间隔 | 描模型路径的时间间隔，单位秒 | Integer | | | 10 |

| modelStreamStartTime | 模型流的起始时间 | 模型流的起始时间。默认从当前时刻开始读。使用yyyy-mm-dd hh:mm:ss.fffffffff格式，详见Timestamp.valueOf(String s) | String | | | null |

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. import time, datetime
  8. import numpy as np
  9. import pandas as pd
  11. data = pd.DataFrame([
  12.     [0, datetime.datetime.fromisoformat("2021-11-01 00:00:00"), 100.0],
  13.     [0, datetime.datetime.fromisoformat("2021-11-02 00:00:00"), 200.0],
  14.     [0, datetime.datetime.fromisoformat("2021-11-03 00:00:00"), 300.0],
  15.     [0, datetime.datetime.fromisoformat("2021-11-04 00:00:00"), 400.0],
  16.     [0, datetime.datetime.fromisoformat("2021-11-06 00:00:00"), 500.0],
  17.     [0, datetime.datetime.fromisoformat("2021-11-07 00:00:00"), 600.0],
  18.     [0, datetime.datetime.fromisoformat("2021-11-08 00:00:00"), 700.0],
  19.     [0, datetime.datetime.fromisoformat("2021-11-09 00:00:00"), 800.0],
  20.     [0, datetime.datetime.fromisoformat("2021-11-10 00:00:00"), 900.0],
  21.     [0, datetime.datetime.fromisoformat("2021-11-11 00:00:00"), 800.0],
  22.     [0, datetime.datetime.fromisoformat("2021-11-12 00:00:00"), 700.0],
  23.     [0, datetime.datetime.fromisoformat("2021-11-13 00:00:00"), 600.0],
  24.     [0, datetime.datetime.fromisoformat("2021-11-14 00:00:00"), 500.0],
  25.     [0, datetime.datetime.fromisoformat("2021-11-15 00:00:00"), 400.0],
  26.     [0, datetime.datetime.fromisoformat("2021-11-16 00:00:00"), 300.0],
  27.     [0, datetime.datetime.fromisoformat("2021-11-17 00:00:00"), 200.0],
  28.     [0, datetime.datetime.fromisoformat("2021-11-18 00:00:00"), 100.0],
  29.     [0, datetime.datetime.fromisoformat("2021-11-19 00:00:00"), 200.0],
  30.     [0, datetime.datetime.fromisoformat("2021-11-20 00:00:00"), 300.0],
  31.     [0, datetime.datetime.fromisoformat("2021-11-21 00:00:00"), 400.0],
  32.     [0, datetime.datetime.fromisoformat("2021-11-22 00:00:00"), 500.0],
  33.     [0, datetime.datetime.fromisoformat("2021-11-23 00:00:00"), 600.0],
  34.     [0, datetime.datetime.fromisoformat("2021-11-24 00:00:00"), 700.0],
  35.     [0, datetime.datetime.fromisoformat("2021-11-25 00:00:00"), 800.0],
  36.     [0, datetime.datetime.fromisoformat("2021-11-26 00:00:00"), 900.0],
  37.     [0, datetime.datetime.fromisoformat("2021-11-27 00:00:00"), 800.0],
  38.     [0, datetime.datetime.fromisoformat("2021-11-28 00:00:00"), 700.0],
  39.     [0, datetime.datetime.fromisoformat("2021-11-29 00:00:00"), 600.0],
  40.     [0, datetime.datetime.fromisoformat("2021-11-30 00:00:00"), 500.0],
  41.     [0, datetime.datetime.fromisoformat("2021-12-01 00:00:00"), 400.0],
  42.     [0, datetime.datetime.fromisoformat("2021-12-02 00:00:00"), 300.0],
  43.     [0, datetime.datetime.fromisoformat("2021-12-03 00:00:00"), 200.0]
  44. ])
  46. batch_source = dataframeToOperator(data, schemaStr='id int, ts timestamp, series double', op_type='batch')
  48. stream_source = dataframeToOperator(data, schemaStr='id int, ts timestamp, series double', op_type='stream')
  50. lstNetTrainBatchOp = LSTNetTrainBatchOp()\
  51.     .setTimeCol("ts")\
  52.     .setSelectedCol("series")\
  53.     .setNumEpochs(10)\
  54.     .setWindow(24)\
  55.     .setHorizon(1)\
  56.     .linkFrom(batch_source)
  58. overCountWindowStreamOp = OverCountWindowStreamOp()\
  59.     .setClause("MTABLE_AGG_PRECEDING(ts, series) as mtable_agg_series")\
  60.     .setTimeCol("ts")\
  61.     .setPrecedingRows(24)
  63. lstNetPredictStreamOp = LSTNetPredictStreamOp(lstNetTrainBatchOp)\
  64.     .setPredictNum(1)\
  65.     .setPredictionCol("pred")\
  66.     .setReservedCols([])\
  67.     .setValueCol("mtable_agg_series")
  69. lstNetPredictStreamOp\
  70.     .linkFrom(
  71.         overCountWindowStreamOp\
  72.         .linkFrom(stream_source)\
  73.         .filter("ts = TO_TIMESTAMP('2021-12-03 00:00:00')")
  74.     )\
  75.     .print();
  77. StreamOperator.execute();

Java 代码

  1. import org.apache.flink.types.Row;
  3. import com.alibaba.alink.operator.batch.BatchOperator;
  4. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  5. import com.alibaba.alink.operator.stream.StreamOperator;
  6. import com.alibaba.alink.operator.stream.feature.OverCountWindowStreamOp;
  7. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  8. import com.alibaba.alink.operator.stream.timeseries.LSTNetPredictStreamOp;
  9. import org.junit.Test;
  11. import java.sql.Timestamp;
  12. import java.util.Arrays;
  13. import java.util.List;
  15. public class LSTNetPredictStreamOpTest {
  17.     @Test
  18.     public void testLSTNetTrainBatchOp() throws Exception {
  19.         BatchOperator.setParallelism(1);
  21.         List <Row> data = Arrays.asList(
  22.             Row.of(0, Timestamp.valueOf("2021-11-01 00:00:00"), 100.0),
  23.             Row.of(0, Timestamp.valueOf("2021-11-02 00:00:00"), 200.0),
  24.             Row.of(0, Timestamp.valueOf("2021-11-03 00:00:00"), 300.0),
  25.             Row.of(0, Timestamp.valueOf("2021-11-04 00:00:00"), 400.0),
  26.             Row.of(0, Timestamp.valueOf("2021-11-06 00:00:00"), 500.0),
  27.             Row.of(0, Timestamp.valueOf("2021-11-07 00:00:00"), 600.0),
  28.             Row.of(0, Timestamp.valueOf("2021-11-08 00:00:00"), 700.0),
  29.             Row.of(0, Timestamp.valueOf("2021-11-09 00:00:00"), 800.0),
  30.             Row.of(0, Timestamp.valueOf("2021-11-10 00:00:00"), 900.0),
  31.             Row.of(0, Timestamp.valueOf("2021-11-11 00:00:00"), 800.0),
  32.             Row.of(0, Timestamp.valueOf("2021-11-12 00:00:00"), 700.0),
  33.             Row.of(0, Timestamp.valueOf("2021-11-13 00:00:00"), 600.0),
  34.             Row.of(0, Timestamp.valueOf("2021-11-14 00:00:00"), 500.0),
  35.             Row.of(0, Timestamp.valueOf("2021-11-15 00:00:00"), 400.0),
  36.             Row.of(0, Timestamp.valueOf("2021-11-16 00:00:00"), 300.0),
  37.             Row.of(0, Timestamp.valueOf("2021-11-17 00:00:00"), 200.0),
  38.             Row.of(0, Timestamp.valueOf("2021-11-18 00:00:00"), 100.0),
  39.             Row.of(0, Timestamp.valueOf("2021-11-19 00:00:00"), 200.0),
  40.             Row.of(0, Timestamp.valueOf("2021-11-20 00:00:00"), 300.0),
  41.             Row.of(0, Timestamp.valueOf("2021-11-21 00:00:00"), 400.0),
  42.             Row.of(0, Timestamp.valueOf("2021-11-22 00:00:00"), 500.0),
  43.             Row.of(0, Timestamp.valueOf("2021-11-23 00:00:00"), 600.0),
  44.             Row.of(0, Timestamp.valueOf("2021-11-24 00:00:00"), 700.0),
  45.             Row.of(0, Timestamp.valueOf("2021-11-25 00:00:00"), 800.0),
  46.             Row.of(0, Timestamp.valueOf("2021-11-26 00:00:00"), 900.0),
  47.             Row.of(0, Timestamp.valueOf("2021-11-27 00:00:00"), 800.0),
  48.             Row.of(0, Timestamp.valueOf("2021-11-28 00:00:00"), 700.0),
  49.             Row.of(0, Timestamp.valueOf("2021-11-29 00:00:00"), 600.0),
  50.             Row.of(0, Timestamp.valueOf("2021-11-30 00:00:00"), 500.0),
  51.             Row.of(0, Timestamp.valueOf("2021-12-01 00:00:00"), 400.0),
  52.             Row.of(0, Timestamp.valueOf("2021-12-02 00:00:00"), 300.0),
  53.             Row.of(0, Timestamp.valueOf("2021-12-03 00:00:00"), 200.0)
  54.         );
  56.         MemSourceBatchOp memSourceBatchOp = new MemSourceBatchOp(
  57.             data, "id int, ts timestamp, series double"
  58.         );
  60.         MemSourceStreamOp memSourceStreamOp = new MemSourceStreamOp(
  61.             data, "id int, ts timestamp, series double"
  62.         );
  64.         LSTNetTrainBatchOp lstNetTrainBatchOp = new LSTNetTrainBatchOp()
  65.             .setTimeCol("ts")
  66.             .setSelectedCol("series")
  67.             .setNumEpochs(10)
  68.             .setWindow(24)
  69.             .setHorizon(1)
  70.             .linkFrom(memSourceBatchOp);
  72.         OverCountWindowStreamOp overCountWindowStreamOp = new OverCountWindowStreamOp()
  73.             .setClause("MTABLE_AGG_PRECEDING(ts, series) as mtable_agg_series")
  74.             .setTimeCol("ts")
  75.             .setPrecedingRows(24);
  77.         LSTNetPredictStreamOp lstNetPredictStreamOp = new LSTNetPredictStreamOp(lstNetTrainBatchOp)
  78.             .setPredictNum(1)
  79.             .setPredictionCol("pred")
  80.             .setReservedCols()
  81.             .setValueCol("mtable_agg_series");
  83.         lstNetPredictStreamOp
  84.             .linkFrom(
  85.                 overCountWindowStreamOp
  86.                     .linkFrom(memSourceStreamOp)
  87.                     .filter("ts = TO_TIMESTAMP('2021-12-03 00:00:00')")
  88.             )
  89.             .print();
  91.         StreamOperator.execute();
  92.     }
  94. }

运行结果

| pred | | —- |

| {“data”:{“ts”:[“2021-12-04 00:00:00.0”],”series”:[441.76019287109375]},”schema”:”ts TIMESTAMP,series DOUBLE”} |