Scala Spark - Для цикла в рамке данных и сравнить даты

Question

У меня есть кадр данных, который имеет 3 колонки, как это:

--------------------------------------------- 
| x(string) | date(date) | value(int) | 
--------------------------------------------- 

Я хочу, чтобы выбрать все строки [I], которые удовлетворяют все 4 условия:

1) строки [I] и строки [я - 1] имеют одинаковое значение в столбце 'х'

и

2) 'дата' в строке [я] == 'дата' в строке [i - 1] + 1 (два дня подряд)

И

3) 'значение' в строке [я]> 5

И

4) 'значение' в строке [я - 1] < = 5

Я думаю, мне нужен цикл For, но не знаю, как именно! Пожалуйста, помогите мне!

Всякая помощь очень ценится!

Answer 1

Это может быть очень легко сделать с помощью функции окна, обратите внимание на отставание function:

import org.apache.spark.sql.types._ 
import org.apache.spark.sql._ 
import sqlContext.implicits._ 
import org.apache.spark.sql.functions._ 
import org.apache.spark.sql.expressions._ 

// test data 
val list = Seq(
    ("x", "2016-12-13", 1), 
    ("x", "2016-12-14", 7) 
    ); 
val df = sc.parallelize(list).toDF("x", "date", "value"); 

// add lags - so read previous value from dataset 
val withPrevs = df 
    .withColumn ("prevX", lag('x, 1).over(Window.orderBy($"date"))) 
    .withColumn ("prevDate", lag('date, 1).over(Window.orderBy($"date"))) 
    .withColumn ("prevValue", lag('value, 1).over(Window.orderBy($"date"))) 

// filter values and select only needed fields 
withPrevs 
    .where('x === 'prevX) 
    .where('value > lit(5)) 
    .where('prevValue < lit(5)) 
    .where('date === date_add('prevDate, 1)) 
    .select('x, 'date, 'value) 
    .show() 

Примечание, что без порядка, то есть на сегодняшний день, это не может быть сделано. Dataset не имеет никакого значимого порядка, вы должны указать порядок explicity

Answer 2

Если у вас есть DataFrame, тогда все, что вам нужно сделать, это позвонить в фильтр на DataFrame, все ваши условия.

Например:

df1.filter($"Column1" === 2 || $"Column2" === 3) 

Вы можете пройти столько условий, сколько вы хотите. Он вернет вам новый DataFrame с filtered data.