сравнить столбец даты с максимальной датой в spark sql

Question

Использование Spark2.3.0 и Scala

Имея таблицу, подобную приведенной ниже:

created_date mth    ColA    
2019-01-01  2019-01 a
2019-01-01  2019-01 b
2019-01-02  2019-01 a
2019-01-02  2019-01 b
.
.
2019-06-26  2019-01 a

И схема выглядит следующим образом:

root
 |-- transaction_created_date: string (nullable = true)
 |-- txn_mth: string (nullable = true)
 |-- ColA: string (nullable = true)

хотите сравнить столбец создал_датус max_date и создайте новый столбец

Попробовал как показано ниже:

var max_date = sparkVal.sql(s"""SELECT cast(max(created_date)                 
        as DATE) from BASE_TABLE""").first()
val maxDateValue = max_date.get(0)
var day_counter=10
val data =spark.sql(s"""SELECT
       created_date,
       mth,
       sum(if(date_add(created_date+$day_counter) > cast($maxDateValue as DATE) ),1,0)) 
       as Total_arrival from BASE_TALE a""")

lets say max_date = 2019-06-29 хочу вывод как

created_date mth    Total_arrival
2019-01-01  2019-01 1
2019-01-01  2019-01 1
2019-01-01  2019-01 1
2019-01-02  2019-01 1
.
.
2019-06-26  2019-01 0
2019-06-27  2019-01 0
2019-06-28  2019-01 0
2019-06-29  2019-01 0
2019-06-30  2019-01 0

getting below error :

org.apache.spark.sql.AnalysisException: невозможно разрешить 'CAST ((((2019 - 6) - 26) AS DATE)' из-за несоответствия типов данных: невозможно привести int к дате;строка 43 поз 106;

Может кто-нибудь помочь с приведением максимального значения, чтобы его можно было сравнить со столбцом даты?

Answer 1

Одна реализация может быть следующей:

object TestSO {

  def main(args: Array[String]) : Unit = {
    // dataset
    implicit val spark: SparkSession =
      SparkSession
        .builder()
        .master("local[1]")
        .appName("Test")
        .getOrCreate()

    import org.apache.spark.sql.functions.{to_date, col, max, when, date_add, lit}

    val data = Seq(Row("2019-01-01", "2019-01", "a"),
                   Row("2019-01-01", "2019-01", "b"),
                   Row("2019-01-02", "2019-01", "a"),
                   Row("2019-01-02", "2019-01", "b"))

    val df = spark.createDataFrame(spark.sparkContext.parallelize(data), StructType(List(StructField("transaction_created_date", StringType, false),
      StructField("txn_mth", StringType, false),
      StructField("ColA", StringType, false))))

    // Add a column with a new column as date. It could be done all in one line
    val df_withdate = df.withColumn("transaction_created_date",
      to_date(col("transaction_created_date")))

    var day_counter=10

    // Getting the max
    val max_date = df_withdate
      .select(max(col("transaction_created_date")))
      .collect()(0)(0)

    // Put 1, in rows where creation_date + day_counter > max_date
    val result_df = df_withdate.withColumn("Total_arrival",
      when(date_add(col("transaction_created_date"), day_counter) > to_date(lit(max_date)), 1)
     .otherwise(0))

    result_df.show()
  }
}

Это дает:

+------------------------+-------+----+-------------+
|transaction_created_date|txn_mth|ColA|Total_arrival|
+------------------------+-------+----+-------------+
|              2019-01-01|2019-01|   a|            1|
|              2019-01-01|2019-01|   b|            1|
|              2019-01-02|2019-01|   a|            1|
|              2019-01-02|2019-01|   b|            1|
+------------------------+-------+----+-------------+