//设置package包名称以及导入依赖的类
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions.max

object DrawdownCalculator {
  def calculate(sparkSession: SparkSession, df: DataFrame): Double = {

    val windowUptoCurrentRow = Window.orderBy("date").rowsBetween(Long.MinValue, 0)
    val dfWithRollingMaxPrice = df.withColumn("rolling_max_price",
                                              max(df("price")).over(windowUptoCurrentRow))

    val dfWithRollingDrawdowns = dfWithRollingMaxPrice.withColumn("rolling_dd",
      max(dfWithRollingMaxPrice("rolling_max_price") - dfWithRollingMaxPrice("price")).over(windowUptoCurrentRow))

    dfWithRollingDrawdowns.createOrReplaceTempView("DrawdownCalculation")

    val dfWithOrderedDrawndowns = sparkSession.sql("SELECT date, price, rolling_dd, rolling_max_price, " +
                                                   "(rolling_dd / rolling_max_price) as drawdown_pct " +
                                                   "FROM DrawdownCalculation ORDER BY drawdown_pct ASC")

    dfWithOrderedDrawndowns.show()

    val rollingDrawdown = dfWithOrderedDrawndowns.first().getDouble(2)
    val rollingMaxPrice = dfWithOrderedDrawndowns.first().getDouble(3)
    val maxDrawdownPct = dfWithOrderedDrawndowns.first().getDouble(4)

    maxDrawdownPct
  }
} 

//设置package包名称以及导入依赖的类
package org.alghimo.spark.dimensionalModelling

import org.apache.spark.sql._
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{lit, max, row_number}


  def naturalKeyColumns: Seq[String]

  def enrichedDimensionsToDimensions(enrichedDims: EnrichedDimensions, refreshDimensionTable: Boolean = false): Dimensions = {
    val dimensions = dimensionTable(refreshDimensionTable)
    val maxSk = dimensions.select(max(surrogateKeyColumn)).as[Long].collect().head
    val rankWindow = Window.partitionBy().orderBy(naturalKeyColumns.head, naturalKeyColumns.tail:_*)

    enrichedDims
      .withColumn(surrogateKeyColumnName, lit(maxSk) + (row_number() over rankWindow))
      .withColumn(startTimestampColumnName, timestampColumn)
      .withColumn(endTimestampColumnName, lit(null).cast("timestamp"))
      .withColumn(isCurrentColumnName, lit(true))
      .selectExpr(dimensions.columns:_*)
      .as[DIM]
  }

  def keepOnlyMostRecentEvents(enrichedDimensions: EnrichedDimensions): EnrichedDimensions = {
    val naturalKeyWindow = Window.partitionBy(naturalKeyColumns.map(new Column(_)):_*).orderBy(timestampColumn.desc)

    enrichedDimensions
      .withColumn("row_num", row_number() over naturalKeyWindow)
      .filter("row_num = 1")
      .drop("row_num")
      .as[ENRICHED_DIM]
  }
} 

//设置package包名称以及导入依赖的类
package com.jjzhk.sparkexamples.sql.search

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._


object SparkSQLSearch {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("SparkSQLSearch").master("spark://Master:7077").enableHiveSupport().getOrCreate()
    import spark.implicits._
    spark.sql("use hive")
    var dfuv = spark.sql("select date, item, count(*) as uv from searchinfo group by date, item order by date")
      .withColumn("number", row_number().over(Window.partitionBy($"date").orderBy($"date", $"uv".desc)))
          .filter($"number".leq(5))
    var dfitems = spark.sql("select * from items")
    dfuv.join(dfitems, Seq("item")).select("date", "itemname", "uv").show(200)
  }
}