問(wèn)題描述
我有以下數(shù)據(jù).
+----------+----+-------+-----------------------+
| date|item|avg_val|conditions |
+----------+----+-------+-----------------------+
|01-10-2020| x| 10| 0|
|02-10-2020| x| 10| 0|
|03-10-2020| x| 15| 1|
|04-10-2020| x| 15| 1|
|05-10-2020| x| 5| 0|
|06-10-2020| x| 13| 1|
|07-10-2020| x| 10| 1|
|08-10-2020| x| 10| 0|
|09-10-2020| x| 15| 1|
|01-10-2020| y| 10| 0|
|02-10-2020| y| 18| 0|
|03-10-2020| y| 6| 1|
|04-10-2020| y| 10| 0|
|05-10-2020| y| 20| 0|
+----------+----+-------+-----------------------+
我正在嘗試基于
- 如果標(biāo)志值為 0，則新列值將為 0.
- 如果標(biāo)志為 1，則新列將為 1，接下來(lái)的四個(gè) N 行數(shù)將為零，即無(wú)需檢查下一個(gè) N 值.此過(guò)程將應(yīng)用于每個(gè)項(xiàng)目，即按項(xiàng)目分區(qū)將起作用.
我在這里使用了 N = 4，
I have used here N = 4,
我已經(jīng)使用了下面的代碼，但沒(méi)有有效的窗口函數(shù)是否有任何優(yōu)化的方法.
I have used the below code but not effienntly windowing function is there any optimized way.
DROP TEMPORARY TABLE t2;
CREATE TEMPORARY TABLE t2
SELECT *,
MAX(conditions) OVER (PARTITION BY item ORDER BY item,`date` ROWS 4 PRECEDING ) AS new_row
FROM record
ORDER BY item,`date`;
DROP TEMPORARY TABLE t3;
CREATE TEMPORARY TABLE t3
SELECT *,ROW_NUMBER() OVER (PARTITION BY item,new_row ORDER BY item,`date`) AS e FROM t2;
SELECT *,CASE WHEN new_row=1 AND e%5>1 THEN 0
WHEN new_row=1 AND e%5=1 THEN 1 ELSE 0 END AS flag FROM t3;
輸出類似于
+----------+----+-------+-----------------------+-----+
| date|item|avg_val|conditions |flag |
+----------+----+-------+-----------------------+-----+
|01-10-2020| x| 10| 0| 0|
|02-10-2020| x| 10| 0| 0|
|03-10-2020| x| 15| 1| 1|
|04-10-2020| x| 15| 1| 0|
|05-10-2020| x| 5| 0| 0|
|06-10-2020| x| 13| 1| 0|
|07-10-2020| x| 10| 1| 0|
|08-10-2020| x| 10| 0| 0|
|09-10-2020| x| 15| 1| 1|
|01-10-2020| y| 10| 0| 0|
|02-10-2020| y| 18| 0| 0|
|03-10-2020| y| 6| 1| 1|
|04-10-2020| y| 10| 0| 0|
|05-10-2020| y| 20| 0| 0|
+----------+----+-------+-----------------------+-----+
但是我無(wú)法獲得輸出，我嘗試了更多.
But i am unable to get the ouput , i have tried more.
推薦答案
正如評(píng)論中所建議的(@nbk 和 @Akina)，您將需要某種迭代器來(lái)實(shí)現(xiàn)邏輯.對(duì)于 SparkSQL 和 Spark 2.4+ 版，我們可以使用內(nèi)置函數(shù) aggregate 并設(shè)置一個(gè)結(jié)構(gòu)數(shù)組和一個(gè)計(jì)數(shù)器作為累加器.下面是一個(gè)名為 record 的示例數(shù)據(jù)框和表(假設(shè) conditions 列中的值為 0 或 1):
As suggested in the comments(by @nbk and @Akina), you will need some sort of iterator to implement the logic. With SparkSQL and Spark version 2.4+, we can use the builtin function aggregate and set an array of structs plus a counter as the accumulator. Below is an example dataframe and table named record(assume values in conditions column are either 0 or 1):
val df = Seq(
("01-10-2020", "x", 10, 0), ("02-10-2020", "x", 10, 0), ("03-10-2020", "x", 15, 1),
("04-10-2020", "x", 15, 1), ("05-10-2020", "x", 5, 0), ("06-10-2020", "x", 13, 1),
("07-10-2020", "x", 10, 1), ("08-10-2020", "x", 10, 0), ("09-10-2020", "x", 15, 1),
("01-10-2020", "y", 10, 0), ("02-10-2020", "y", 18, 0), ("03-10-2020", "y", 6, 1),
("04-10-2020", "y", 10, 0), ("05-10-2020", "y", 20, 0)
).toDF("date", "item", "avg_val", "conditions")
df.createOrReplaceTempView("record")
SQL:
spark.sql("""
SELECT t1.item, m.*
FROM (
SELECT item,
sort_array(collect_list(struct(date,avg_val,int(conditions) as conditions,conditions as flag))) as dta
FROM record
GROUP BY item
) as t1 LATERAL VIEW OUTER inline(
aggregate(
/* expr: set up array `dta` from the 2nd element to the last
* notice that indices for slice function is 1-based, dta[i] is 0-based
*/
slice(dta,2,size(dta)),
/* start: set up and initialize `acc` to a struct containing two fields:
* - dta: an array of structs with a single element dta[0]
* - counter: number of rows after flag=1, can be from `0` to `N+1`
*/
(array(dta[0]) as dta, dta[0].conditions as counter),
/* merge: iterate through the `expr` using x and update two fields of `acc`
* - dta: append values from x to acc.dta array using concat + array functions
* update flag using `IF(acc.counter IN (0,5) and x.conditions = 1, 1, 0)`
* - counter: increment by 1 if acc.counter is between 1 and 4
* , otherwise set value to x.conditions
*/
(acc, x) -> named_struct(
'dta', concat(acc.dta, array(named_struct(
'date', x.date,
'avg_val', x.avg_val,
'conditions', x.conditions,
'flag', IF(acc.counter IN (0,5) and x.conditions = 1, 1, 0)
))),
'counter', IF(acc.counter > 0 and acc.counter < 5, acc.counter+1, x.conditions)
),
/* finish: retrieve acc.dta only and discard acc.counter */
acc -> acc.dta
)
) m
""").show(50)
結(jié)果:
+----+----------+-------+----------+----+
|item| date|avg_val|conditions|flag|
+----+----------+-------+----------+----+
| x|01-10-2020| 10| 0| 0|
| x|02-10-2020| 10| 0| 0|
| x|03-10-2020| 15| 1| 1|
| x|04-10-2020| 15| 1| 0|
| x|05-10-2020| 5| 0| 0|
| x|06-10-2020| 13| 1| 0|
| x|07-10-2020| 10| 1| 0|
| x|08-10-2020| 10| 0| 0|
| x|09-10-2020| 15| 1| 1|
| y|01-10-2020| 10| 0| 0|
| y|02-10-2020| 18| 0| 0|
| y|03-10-2020| 6| 1| 1|
| y|04-10-2020| 10| 0| 0|
| y|05-10-2020| 20| 0| 0|
+----+----------+-------+----------+----+
地點(diǎn):
- 使用
groupby將同一項(xiàng)目的行收集到名為 dta 列的結(jié)構(gòu)數(shù)組中，該列具有 4 個(gè)字段:date、avg_val、conditions 和 flag 并按 date 排序 - 使用
aggregate函數(shù)遍歷上述結(jié)構(gòu)體數(shù)組，根據(jù)counter和conditions更新flag字段strong>(詳情見上面SQL代碼注釋) - 使用
Lateral VIEW和 inline 函數(shù)分解來(lái)自聚合函數(shù)的結(jié)果結(jié)構(gòu)數(shù)組
- use
groupbyto collect rows for the same item into an array of structs named dta column with 4 fields: date, avg_val, conditions and flag and sorted by date - use
aggregatefunction to iterate through the above array of structs, update the flag field based on counter and conditions (details see the above SQL code comments) - use
Lateral VIEWand inline function to explode the resulting array of structs from the aggregate function
注意事項(xiàng):
(1) 建議的 SQL 適用于 N=4，其中我們有 acc.counter IN (0,5) 和 acc.counter <;5 在 SQL 中.對(duì)于任何 N，將上述調(diào)整為:acc.counter IN (0,N+1) 和 acc.counter <;N+1，下圖為N=2 相同樣本數(shù)據(jù)的結(jié)果:
(1) the proposed SQL is for N=4, where we have acc.counter IN (0,5) and acc.counter < 5 in the SQL. For any N, adjust the above to: acc.counter IN (0,N+1) and acc.counter < N+1, the below shows the result for N=2 with the same sample data:
+----+----------+-------+----------+----+
|item| date|avg_val|conditions|flag|
+----+----------+-------+----------+----+
| x|01-10-2020| 10| 0| 0|
| x|02-10-2020| 10| 0| 0|
| x|03-10-2020| 15| 1| 1|
| x|04-10-2020| 15| 1| 0|
| x|05-10-2020| 5| 0| 0|
| x|06-10-2020| 13| 1| 1|
| x|07-10-2020| 10| 1| 0|
| x|08-10-2020| 10| 0| 0|
| x|09-10-2020| 15| 1| 1|
| y|01-10-2020| 10| 0| 0|
| y|02-10-2020| 18| 0| 0|
| y|03-10-2020| 6| 1| 1|
| y|04-10-2020| 10| 0| 0|
| y|05-10-2020| 20| 0| 0|
+----+----------+-------+----------+----+
(2) 我們使用dta[0] 來(lái)初始化acc，它包括其字段的值和數(shù)據(jù)類型.理想情況下，我們應(yīng)該確保這些字段的數(shù)據(jù)類型正確，以便正確進(jìn)行所有計(jì)算.例如在計(jì)算 acc.counter 時(shí)，如果 conditions 是 StringType，acc.counter+1 將返回一個(gè)帶有 DoubleType 值的 StringType
(2) we use dta[0] to initialize acc which includes both the values and datatypes of its fields. Ideally, we should make sure data types of these fields right so that all calculations are correctly conducted. for example when calculating acc.counter, if conditions is StringType, acc.counter+1 will return a StringType with a DoubleType value
spark.sql("select '2'+1").show()
+---------------------------------------+
|(CAST(2 AS DOUBLE) + CAST(1 AS DOUBLE))|
+---------------------------------------+
| 3.0|
+---------------------------------------+
當(dāng)使用 acc.counter IN (0,5) 或 acc.counter 將其值與整數(shù)進(jìn)行比較時(shí)，可能會(huì)產(chǎn)生浮點(diǎn)錯(cuò)誤.5.根據(jù) OP 的反饋，這產(chǎn)生了錯(cuò)誤的結(jié)果，沒(méi)有任何警告/錯(cuò)誤消息.
Which could yield floating-point errors when comparing their value with integers using acc.counter IN (0,5) or acc.counter < 5. Based on OP's feedback, this produced incorrect result without any WARNING/ERROR message.
一種解決方法是在設(shè)置聚合函數(shù)的第二個(gè)參數(shù)時(shí)使用 CAST 指定確切的字段類型，以便在任何類型不匹配時(shí)報(bào)告錯(cuò)誤，見下文:
One workaround is to specify exact field types using CAST when setting up the 2nd argument of aggregate function so it reports ERROR when any types mismatch, see below:
CAST((array(dta[0]), dta[0].conditions) as struct<dta:array<struct<date:string,avg_val:string,conditions:int,flag:int>>,counter:int>),
另一種在創(chuàng)建 dta 列時(shí)強(qiáng)制類型的解決方案，在此示例中，請(qǐng)參閱以下代碼中的 int(conditions) as conditions:
Another solution it to force types when creating dta column, in this example, see int(conditions) as conditions in below code:
SELECT item,
sort_array(collect_list(struct(date,avg_val,int(conditions) as conditions,conditions as flag))) as dta
FROM record
GROUP BY item
我們也可以在計(jì)算中強(qiáng)制使用數(shù)據(jù)類型，例如，參見下面的int(acc.counter+1):
IF(acc.counter > 0 and acc.counter < 5, int(acc.counter+1), x.conditions)
這篇關(guān)于如何有效地使用窗口函數(shù)根據(jù) N 個(gè)先前值來(lái)決定接下來(lái)的 N 個(gè)行的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，也希望大家多多支持html5模板網(wǎng)！