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

功能介绍

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

注意事项

  • 时间序列列，是特殊的MTable类型，一列是时间，一列是值，参考运行结果的data列。
  • 查找的时间在数据列中不存在时，用相邻时刻的值差值得到结果。

    参数说明

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

| 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 double', 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.             LookupValueInTimeSeriesStreamOp()
  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.LookupValueInTimeSeriesBatchOp;
  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 LookupValueInTimeSeriesStreamOpTest {
  17.     @Test
  18.     public void test() throws Exception {
  20.         List <Row> mTableData = Arrays.asList(
  21.             Row.of(new Timestamp(1), 10.0),
  22.             Row.of(new Timestamp(2), 11.0),
  23.             Row.of(new Timestamp(3), 12.0),
  24.             Row.of(new Timestamp(4), 13.0),
  25.             Row.of(new Timestamp(5), 14.0),
  26.             Row.of(new Timestamp(6), 15.0),
  27.             Row.of(new Timestamp(7), 16.0),
  28.             Row.of(new Timestamp(8), 17.0),
  29.             Row.of(new Timestamp(9), 18.0),
  30.             Row.of(new Timestamp(10), 19.0)
  31.         );
  33.         MTable mtable = new MTable(mTableData, "ts timestamp, val double");
  35.         MemSourceStreamOp source = new MemSourceStreamOp(
  36.             new Object[][] {
  37.                 {1, new Timestamp(5), mtable}
  38.             },
  39.             new String[] {"id", "ts", "data"});
  41.         source
  42.             .link(new LookupValueInTimeSeriesStreamOp()
  43.                 .setTimeCol("ts")
  44.                 .setTimeSeriesCol("data")
  45.                 .setOutputCol("out")
  46.             )
  47.             .print();
  49.         StreamOperator.execute();
  50.     }
  51. }

运行结果

| id | ts | data | out | | —- | —- | —- | —- |

| 1 | 1970-01-01 08:00:00.005 | MTable(10,2)(ts,val)
1970-01-01 08:00:00.001 | 10.000
1970-01-01 08:00:00.002 | 11.000
1970-01-01 08:00:00.003 | 12.000
1970-01-01 08:00:00.004 | 13.000
1970-01-01 08:00:00.005 | 14.000 | 14.0000 |