不同的总和和分组依据

Question

我有一个数据集 [附加示例]，我想从中创建 2 个表；

+------+------------+-------+-------+-------+--------+
| corp |  product   | data  | Group | sales | market |
+------+------------+-------+-------+-------+--------+
| A    | Eli        | 43831 | A     |   100 | I      |
| A    | Eli        | 43831 | B     |   100 | I      |
| B    | Sut        | 43831 | A     |    80 | I      |
| A    | Api        | 43831 | C     |    50 | C or D |
| A    | Api        | 43831 | D     |    50 | C or D |
| B    | Konkurent2 | 43831 | C     |    40 | C or D |
+------+------------+-------+-------+-------+--------+

1st - 按市场求和（销售额）并排除重复的行，所以我想在特定日期范围（数据列）中得到每个市场的销售额，但排除重复的 - 我有它们，因为 1 个产品可以超过1组

所以第一个表，例如，对于 MRCC I，看起来像：

+--------+-------+-------+
| market | sales | data  |
+--------+-------+-------+
|  I     |   180 | 43831 |
+--------+-------+-------+

然后我希望第二个表看起来像上面的一个，但是在市场和日期中添加一个带有唯一产品名称的“字典”附加列，所以对于 MRCC，它看起来像：

+--------+-------+-------+----------------+
| market | sales | data  | unique product |
+--------+-------+-------+----------------+
| I      |   180 | 43831 | eli            |
| I      |   180 | 43831 | Sut            |
+--------+-------+-------+----------------+

问题是，我在 SQL 方面没有那么丰富的经验，而且我对 DataProcessing 还很陌生，我正在使用的系统允许我通过一些“视觉”配方或通过我的 SQL 代码来进行一些数据处理不是那么熟悉。甚至令人困惑的是我可以在 3 SQL DBMS、Impala、Hive、Spark SQL 之间进行选择——例如，我使用 Impala 创建市场列，脚本看起来像这样，我不确定这是否是“纯”Impala 语法：

SELECT * from
(  
   -- mrc I --
    SELECT *,case when 
    (`product`="Eli") 
    or
    (`product`="Sut")  
    THEN  "MRCC I"
    end as market
    FROM x.`y`  
  )a
where market is not null

你能给我一些关于代码结构的提示吗？如果可能的话？

谢谢， em

Answer 1

  import spark.implicits._
  import org.apache.spark.sql.functions._

  case class Sale(
                 corp:    String,
                 product: String,
                 data:    Long,
                 group:   String,
                 sales:   Long,
                 market:  String
                 )

  val df = Seq(
    Sale("A", "Eli", 43831, "A", 100, "I"),
    Sale("A", "Eli", 43831, "B", 100, "I"),
    Sale("A", "Sut", 43831, "A", 80, "I"),
    Sale("A", "Api", 43831, "C", 50, "C or D"),
    Sale("A", "Api", 43831, "D", 50, "C or D"),
    Sale("B", "Konkurent2", 43831, "C", 40, "C or D")
  ).toDF()

  val t2 = df.dropDuplicates(Seq("corp", "product", "data", "market"))
    .groupBy("market", "product", "data").sum("sales")
    .select(
      'market,
      col("sum(sales)").alias("sales"),
      'data,
      'product.alias("unique product")
    )

  t2.show(false)
//  +------+-----+-----+--------------+
//  |market|sales|data |unique product|
//  +------+-----+-----+--------------+
//  |I     |80   |43831|Sut           |
//  |I     |100  |43831|Eli           |
//  |C or D|40   |43831|Konkurent2    |
//  |C or D|50   |43831|Api           |
//  +------+-----+-----+--------------+

  val t1 = t2.drop("unique product")
    .groupBy("market", "data").sum("sales")
    .select(
      'market,
      col("sum(sales)").alias("sales"),
      'data)

  t1.show(false)
//  +------+-----+-----+
//  |market|sales|data |
//  +------+-----+-----+
//  |I     |180  |43831|
//  |C or D|90   |43831|
//  +------+-----+-----+