标签: window-functions

好吧,最初这只是我和我的一个朋友开玩笑,但它变成了有趣的技术问题:)

我有下stuff表:

CREATE TABLE stuff
(
    id serial PRIMARY KEY,
    volume integer NOT NULL DEFAULT 0,
    priority smallint NOT NULL DEFAULT 0,
);

该表包含我所有内容的记录,包括各自的数量和优先级(我需要多少).

比方说,我有一个指定体积的包1000.我想从表格中选择我可以放入包中的所有东西,首先包装最重要的东西.

这似乎是使用窗口函数的情况,所以这是我提出的查询:

select s.*, sum(volume) OVER previous_rows as total
 from stuff s
 where total < 1000
 WINDOW previous_rows as
  (ORDER BY priority desc ROWS between UNBOUNDED PRECEDING and CURRENT ROW)
 order by priority desc

然而,它的问题在于Postgres抱怨:

ERROR:  column "total" does not exist
LINE 3:  where total < 1000

如果我删除了这个过滤器,总列得到了正确的计算,结果排序正确,但所有的东西都被选中,这不是我想要的.

那么,我该怎么做？如何仅选择可放入包中的物品？

我已经多次找到了以下获取音频信号功率谱的指南:

• 收集N个样本,其中N是2的幂
• 对样本应用合适的窗函数,例如汉宁
• 将窗口样本传递给FFT例程 - 理想情况下,您需要一个实际到复数的FFT,但如果您只有一个是复杂到复杂的FFT,则为所有虚拟输入部分传递0
• 计算FFT输出箱的平方幅度(re*re + im*im)
• (可选)计算每个幅度平方输出仓的10*log10,得到以dB为单位的幅度值
• 现在你已经拥有了你的功率谱,你只需要确定峰值,如果你有一个合理的信噪比,这应该非常简单.注意,频率分辨率随着N的增大而提高.对于上述44.1kHz采样率和N = 32768的示例,每个箱的频率分辨率为44100/32768 = 1.35Hz.

但是......为什么我需要对样本应用窗口函数？这到底意味着什么？

功率谱怎么样,它是采样率范围内每个频率的功率？(例如:windows media player声音可视化器？)

来自PySpark文档rangeBetween:

rangeBetween(start, end)

定义从起点(包括)到结束(包括)的帧边界.

start和end都是当前行的相对值.例如,"0"表示"当前行",而"-1"表示当前行之前的一个关闭,"5"表示当前行之后的五个关闭.

参数:

• 开始 - 边界开始,包括.如果这是-sys.maxsize(或更低),则该框架是无界的.
• 结束 - 边界端,包括端点.如果这是sys.maxsize(或更高),则该框架是无界的.版本1.4中的新功能.

而 rowsBetween

rowsBetween(start, end)

定义从起点(包括)到结束(包括)的帧边界.

start和end都是当前行的相对位置.例如,"0"表示"当前行",而"-1"表示当前行之前的行,"5"表示当前行之后的第五行.

参数:

• 开始 - 边界开始,包括.如果这是-sys.maxsize(或更低),则该框架是无界的.
• 结束 - 边界端,包括端点.如果这是sys.maxsize(或更高),则该框架是无界的.版本1.4中的新功能.

例如rangeBetween,"1 off"与"1行"有何不同？

我已经row_number() partitionBy使用Window在Spark中成功创建了一个by,但是希望通过降序对其进行排序,而不是默认的升序.这是我的工作代码:

from pyspark import HiveContext
from pyspark.sql.types import *
from pyspark.sql import Row, functions as F
from pyspark.sql.window import Window

data_cooccur.select("driver", "also_item", "unit_count", 
    F.rowNumber().over(Window.partitionBy("driver").orderBy("unit_count")).alias("rowNum")).show()

这给了我这个结果:

 +------+---------+----------+------+
 |driver|also_item|unit_count|rowNum|
 +------+---------+----------+------+
 |   s10|      s11|         1|     1|
 |   s10|      s13|         1|     2|
 |   s10|      s17|         1|     3|

在这里我添加desc()以降序:

data_cooccur.select("driver", "also_item", "unit_count", F.rowNumber().over(Window.partitionBy("driver").orderBy("unit_count").desc()).alias("rowNum")).show()

并得到此错误:

AttributeError:'WindowSpec'对象没有属性'desc'

我在这做错了什么？

此答案显示如何从股票代码生成高/低/开/关值:
检索任意时间间隔的聚合

我正在尝试实现基于此的解决方案(PG 9.2),但我很难获得正确的值first_value().

到目前为止,我尝试了两个查询:

SELECT  
    cstamp,
    price,
    date_trunc('hour',cstamp) AS h,
    floor(EXTRACT(minute FROM cstamp) / 5) AS m5,
    min(price) OVER w,
    max(price) OVER w,
    first_value(price) OVER w,
    last_value(price) OVER w
FROM trades
Where date_trunc('hour',cstamp) = timestamp '2013-03-29 09:00:00'
WINDOW w AS (
    PARTITION BY date_trunc('hour',cstamp), floor(extract(minute FROM cstamp) / 5)
    ORDER BY date_trunc('hour',cstamp) ASC, floor(extract(minute FROM cstamp) / 5) ASC
    )
ORDER BY cstamp;

这是结果的一部分:

        cstamp         price      h                 m5  min      max      first    last
"2013-03-29 09:19:14";77.00000;"2013-03-29 09:00:00";3;77.00000;77.00000;77.00000;77.00000

"2013-03-29 09:26:18";77.00000;"2013-03-29 …

是否可以结合窗口函数计算不同的值,如OVER(PARTITION BY id)？目前我的查询如下:

SELECT congestion.date, congestion.week_nb, congestion.id_congestion,
   congestion.id_element,
ROW_NUMBER() OVER(
    PARTITION BY congestion.id_element
    ORDER BY congestion.date),
COUNT(DISTINCT congestion.week_nb) OVER(
    PARTITION BY congestion.id_element
) AS week_count
FROM congestion
WHERE congestion.date >= '2014.01.01'
AND congestion.date <= '2014.12.31'
ORDER BY id_element, date

但是,当我尝试执行查询时,我收到以下错误:

"COUNT(DISTINCT": "DISTINCT is not implemented for window functions"

我正在使用pySpark,并设置了我的数据框,其中两列代表每日资产价格,如下所示:

ind = sc.parallelize(range(1,5))
prices = sc.parallelize([33.3,31.1,51.2,21.3])
data = ind.zip(prices)
df = sqlCtx.createDataFrame(data,["day","price"])

我开始申请df.show():

+---+-----+
|day|price|
+---+-----+
|  1| 33.3|
|  2| 31.1|
|  3| 51.2|
|  4| 21.3|
+---+-----+

哪个好,一切都好.我想有另一个列,其中包含价格列的日常回报,即类似的内容

(price(day2)-price(day1))/(price(day1))

经过大量研究后,我被告知通过应用这些pyspark.sql.window功能可以最有效地完成,但我无法看到.

我正在使用SQL Server 2008 R2,尝试计算移动平均值.对于我视图中的每条记录,我想收集250条先前记录的值,然后计算此选择的平均值.

我的视图列如下:

TransactionID | TimeStamp           | Value | MovAvg
----------------------------------------------------
            1 | 01.09.2014 10:00:12 |     5 |    
            2 | 01.09.2014 10:05:34 |     3 | 
...
          300 | 03.09.2014 09:00:23 |     4 | 

TransactionID是独特的.对于每一个TransactionID,我想计算列值的平均值,超过之前的250个记录.因此,对于TransactionID 300,收集前250行的所有值(视图按TransactionID降序排序),然后在MovAvg列中写入这些值的平均值的结果.我希望收集一系列记录中的数据.

我想通过使用EF在Row_number上分区加载数据.

    SELECT *
    FROM (
       SELECT sf.SerialFlowsId
                     ,sf.GoodsSerialId
                     ,d.FormTypeId
                     , d.GoodsId
                     ,ROW_NUMBER() OVER (PARTITION BY d.GoodsId, sf.GoodsSerialId ORDER BY sf.Date DESC)row
       FROM sam.SerialFlows sf
       INNER JOIN sam.Detail d ON d.DetailId = sf.DetailId
       )z
WHERE z.row =1 
       AND z.FormTypeId=7
       AND z.GoodsId=51532

这个问题是我的期待.

我尝试使用这个表达式,但遗憾的是Zip扩展方法无法在ef中识别

var goodsSerials = context.SerialFlows.OrderByDescending(x => x.Date).GroupBy(x => new { x.Detail.GoodsID, x.Date })
                    .Select(g => new {g})
                    .SelectMany(z => z.g.Select(c => c)).Zip(m, (j, i) => new { GoodSerial=j,j.Detail.FormTypeID,j.Detail.GoodsID,rn=i })
                    .Where(x => x.rn== 1 && x.GoodsID== goodsId && x.FormTypeID==7).Select(x => …

我的问题是由计算spark数据帧中连续行之间差异的用例触发的.

例如,我有:

>>> df.show()
+-----+----------+
|index|      col1|
+-----+----------+
|  0.0|0.58734024|
|  1.0|0.67304325|
|  2.0|0.85154736|
|  3.0| 0.5449719|
+-----+----------+

如果我选择使用"Window"函数计算它们,那么我可以这样做:

>>> winSpec = Window.partitionBy(df.index >= 0).orderBy(df.index.asc())
>>> import pyspark.sql.functions as f
>>> df.withColumn('diffs_col1', f.lag(df.col1, -1).over(winSpec) - df.col1).show()
+-----+----------+-----------+
|index|      col1| diffs_col1|
+-----+----------+-----------+
|  0.0|0.58734024|0.085703015|
|  1.0|0.67304325| 0.17850411|
|  2.0|0.85154736|-0.30657548|
|  3.0| 0.5449719|       null|
+-----+----------+-----------+

问题:我在一个分区中明确地划分了数据帧.这会对性能产生什么影响,如果存在,为什么会这样,我怎么能避免它呢？因为当我没有指定分区时,我收到以下警告:

16/12/24 13:52:27 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.