使用SPARK重新格式化/移动时间序列数据的有效方法

Question

我想使用Spark构建一些时间序列模型。第一步是将序列数据重新格式化为训练样本。这个想法是：

原始顺序数据（每个t*是一个数字）

t1  t2  t3  t4  t5  t6  t7  t8  t9  t10

所需的输出

t1  t2  t3  t4  t5  t6
t2  t3  t4  t5  t6  t7
t3  t4  t5  t6  t7  t8
..................

如何在火花中写一个函数来执行此操作。功能签名应像

重构（数组[Integer]，n：Integer）

返回类型是dataframe或vector

===========我在Spark 1.6.1 =====================

val arraydata=Array[Double](1,2,3,4,5,6,7,8,9,10)
val slideddata = arraydata.sliding(4).toSeq
val rows = arraydata.sliding(4).map{x=>Row(x:_*)}
sc.parallelize(arraydata.sliding(4).toSeq).toDF("Values")

最后一行无法通过错误：

Error:(52, 48) value toDF is not a member of org.apache.spark.rdd.RDD[Array[Double]]
    sc.parallelize(arraydata.sliding(4).toSeq).toDF("Values")

Answer 1

我无法弄清n的重要性，因为它可以用作窗口大小以及必须移动的值。

因此，都有两种口味：

如果n是窗口大小：

def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
sc.parallelize(arrayOfInteger.sliding(shiftValue).toSeq).toDF("values")
}

On REPL:

scala> def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
     | sc.parallelize(arrayOfInteger.sliding(shiftValue).toSeq).toDF("values")
     | }
reformat: (arrayOfInteger: Array[Int], shiftValue: Int)org.apache.spark.sql.DataFrame
scala> val arrayofInteger=(1 to 10).toArray
arrayofInteger: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
scala> reformat(arrayofInteger,3).show
+----------+
|    values|
+----------+
| [1, 2, 3]|
| [2, 3, 4]|
| [3, 4, 5]|
| [4, 5, 6]|
| [5, 6, 7]|
| [6, 7, 8]|
| [7, 8, 9]|
|[8, 9, 10]|
+----------+

如果n是要转移的值：

def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
val slidingValue=arrayOfInteger.size-shiftValue
sc.parallelize(arrayOfInteger.sliding(slidingValue).toSeq).toDF("values")
}

On REPL:

scala> def reformat(arrayOfInteger:Array[Int], shiftValue: Int) ={
     | val slidingValue=arrayOfInteger.size-shiftValue
     | sc.parallelize(arrayOfInteger.sliding(slidingValue).toSeq).toDF("values")
     | }
reformat: (arrayOfInteger: Array[Int], shiftValue: Int)org.apache.spark.sql.DataFrame
scala> val arrayofInteger=(1 to 10).toArray
arrayofInteger: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
scala> reformat(arrayofInteger,3).show(false)
+----------------------+
|values                |
+----------------------+
|[1, 2, 3, 4, 5, 6, 7] |
|[2, 3, 4, 5, 6, 7, 8] |
|[3, 4, 5, 6, 7, 8, 9] |
|[4, 5, 6, 7, 8, 9, 10]|
+----------------------+