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

debugcn 发表于 Dev

塞尔隆

我想使用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
..................

如何在spark中编写函数来做到这一点。功能签名应该像

重新格式化（Array [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")

Shiv4nsh

我无法弄清楚它的重要性，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]|
+----------------------+

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。