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

功能介绍

在时间序列中查找对应时间的值。

参数说明

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

| outputCol | 输出结果列列名 | 输出结果列列名，必选 | String | ✓ | | |

| timeCol | 时间戳列(TimeStamp) | 时间戳列(TimeStamp) | String | ✓ | 所选列类型为 [TIMESTAMP] | |

| timeSeriesCol | 时间序列列 | 时间序列列，是特殊的MTable类型，一列是时间，一列是值 | String | ✓ | 所选列类型为 [M_TABLE] | |

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

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

代码示例

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.             [1,  datetime.datetime.fromtimestamp(1), "10.0"],
  13.             [1,  datetime.datetime.fromtimestamp(2), "11.0"],
  14.             [1,  datetime.datetime.fromtimestamp(3), "12.0"],
  15.             [1,  datetime.datetime.fromtimestamp(4), "13.0"],
  16.             [1,  datetime.datetime.fromtimestamp(5), "14.0"],
  17.             [1,  datetime.datetime.fromtimestamp(6), "15.0"],
  18.             [1,  datetime.datetime.fromtimestamp(7), "16.0"],
  19.             [1,  datetime.datetime.fromtimestamp(8), "17.0"],
  20.             [1,  datetime.datetime.fromtimestamp(9), "18.0"],
  21.             [1,  datetime.datetime.fromtimestamp(10), "19.0"]
  22. ])
  24. source = dataframeToOperator(data, schemaStr='id int, ts timestamp, val string', op_type='stream')
  26. source.link(
  27.             OverCountWindowStreamOp()
  28.                 .setGroupCols(["id"])
  29.                 .setTimeCol("ts")
  30.                 .setPrecedingRows(5)
  31.                 .setClause("mtable_agg_preceding(ts, val) as data")
  32.         ).link(
  33.             ShiftStreamOp()
  34.                 .setValueCol("data")
  35.                 .setShiftNum(7)
  36.                 .setPredictNum(12)
  37.                 .setPredictionCol("predict")
  38.         ).link(
  39.             LookupVectorInTimeSeriesStreamOp()
  40.                 .setTimeCol("ts")
  41.                 .setTimeSeriesCol("predict")
  42.                 .setOutputCol("out")
  43.         ).print()
  45. StreamOperator.execute()

Java 代码

  1. package com.alibaba.alink.operator.stream.timeseries;
  3. import org.apache.flink.types.Row;
  5. import com.alibaba.alink.common.MTable;
  6. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  7. import com.alibaba.alink.operator.batch.timeseries.LookupVectorInTimeSeriesBatchOp;
  8. import com.alibaba.alink.operator.stream.StreamOperator;
  9. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  10. import org.junit.Test;
  12. import java.sql.Timestamp;
  13. import java.util.Arrays;
  14. import java.util.List;
  16. public class LookupVectorInTimeSeriesStreamOpTest {
  17.     @Test
  18.     public void test() throws Exception {
  20.             List <Row> mTableData = Arrays.asList(
  21.                     Row.of(new Timestamp(1), "10.0 21.0"),
  22.                     Row.of(new Timestamp(2), "11.0 22.0"),
  23.                     Row.of(new Timestamp(3), "12.0 23.0"),
  24.                     Row.of(new Timestamp(4), "13.0 24.0"),
  25.                     Row.of(new Timestamp(5), "14.0 25.0"),
  26.                     Row.of(new Timestamp(6), "15.0 26.0"),
  27.                     Row.of(new Timestamp(7), "16.0 27.0"),
  28.                     Row.of(new Timestamp(8), "17.0 28.0"),
  29.                     Row.of(new Timestamp(9), "18.0 29.0"),
  30.                     Row.of(new Timestamp(10), "19.0 30.0")
  31.                 );
  33.                 MemSourceStreamOp source = new MemSourceStreamOp(mTableData, new String[] {"ts", "val"});
  35.                 source.link(
  36.                     new OverCountWindowStreamOp()
  37.                         .setTimeCol("ts")
  38.                         .setPrecedingRows(5)
  39.                         .setClause("mtable_agg(ts, val) as data")
  40.                 ).link(
  41.                     new ShiftStreamOp()
  42.                         .setValueCol("data")
  43.                         .setShiftNum(7)
  44.                         .setPredictNum(12)
  45.                         .setPredictionCol("predict")
  46.                 ).link(
  47.                     new LookupVectorInTimeSeriesStreamOp()
  48.                         .setTimeCol("ts")
  49.                         .setTimeSeriesCol("predict")
  50.                         .setOutputCol("out")
  51.                         .setReservedCols("ts")
  52.                 ).print();
  54.                 StreamOperator.execute();
  55.     }
  56. }

运行结果

| ts | out | | —- | —- |

| 1970-01-01 08:00:00.001 | null |

| 1970-01-01 08:00:00.003 | 10.0 21.0 |

| 1970-01-01 08:00:00.002 | 10.0 21.0 |

| 1970-01-01 08:00:00.005 | 10.0 21.0 |

| 1970-01-01 08:00:00.004 | 10.0 21.0 |

| 1970-01-01 08:00:00.007 | 11.0 22.0 |

| 1970-01-01 08:00:00.006 | 10.0 21.0 |

| 1970-01-01 08:00:00.009 | 13.0 24.0 |

| 1970-01-01 08:00:00.008 | 12.0 23.0 |

| 1970-01-01 08:00:00.01 | 14.0 25.0 |