sparkDataframe中的配置单元嵌套枢轴

jrcvhitl  于 2021-06-27  发布在  Hive
关注(0)|答案(3)|浏览(349)

这个问题在这里已经有答案了

如何透视sparkDataframe(10个答案)
一年前关门了。
我有两个以下的df
主数据框

numberdf(使用配置单元加载创建)

期望输出:

要填充的逻辑
对于field1,需要选择sch\u id,其中cat='pay'和sub\u cat='client'
对于field2,需要在cat='pay'和sub'cat='phr'处选择sch\u id
对于字段3,需要选择pay\u id,其中cat='credit'和sub\u cat='spgrp'
当前在加入之前,我正在对numberdf执行筛选并选取值ex:

masterDF.as("master").join(NumberDF.filter(col("CAT")==="PAY" && col("SUB_CAT")==="phr").as("number"), "$master.id" ==="$number.id" , "leftouter" )
 .select($"master.*", $"number.sch_id".as("field1") )

上述方法需要多个连接。我研究了pivot函数,但它确实解决了我的问题
注意:请忽略代码中的语法错误

3df52oht

3df52oht1#

更好的解决方案是在加入studentdf之前按列(主题)透视dataframe(numberdf)。
pyspark代码如下所示

numberDF = spark.createDataFrame([(1, "Math", 80), (1, "English", 60), (1, "Science", 80)], ["id", "subject", "marks"])
studentDF = spark.createDataFrame([(1, "Vikas")],["id","name"])

>>> numberDF.show()
+---+-------+-----+
| id|subject|marks|
+---+-------+-----+
|  1|   Math|   80|
|  1|English|   60|
|  1|Science|   80|
+---+-------+-----+

>>> studentDF.show()
+---+-----+
| id| name|
+---+-----+
|  1|Vikas|
+---+-----+

pivotNumberDF = numberDF.groupBy("id").pivot("subject").sum("marks")

>>> pivotNumberDF.show()
+---+-------+----+-------+
| id|English|Math|Science|
+---+-------+----+-------+
|  1|     60|  80|     80|
+---+-------+----+-------+

>>> studentDF.join(pivotNumberDF, "id").show()
+---+-----+-------+----+-------+
| id| name|English|Math|Science|
+---+-----+-------+----+-------+
|  1|Vikas|     60|  80|     80|
+---+-----+-------+----+-------+

裁判:http://spark.apache.org/docs/2.4.0/api/python/pyspark.sql.html

6l7fqoea

6l7fqoea2#

最后我用pivot实现了它

flights.groupBy("ID", "CAT")
      .pivot("SUB_CAT", Seq("client", "phr", "spGrp")).agg(avg("SCH_ID").as("SCH_ID"), avg("pay_id").as("pay_id"))
      .groupBy("ID")
      .pivot("CAT", Seq("credit", "price"))
      .agg(
        avg("client_SCH_ID").as("client_sch_id"), avg("client_pay_id").as("client_pay_id")
        , avg("phr_SCH_ID").as("phr_SCH_ID"), avg("phr_pay_id").as("phr_pay_id")
        , avg("spGrp_SCH_ID").as("spGrp_SCH_ID"), avg("spGrp_pay_id").as("spGrp_pay_id")
      )

第一个pivot将返回表

+---+------+-------------+--------------+-----------+------------+-------------+--------------+
| ID|   CAT|client_SCH_ID|client_pay_id |phr_SCH_ID |phr_pay_id  |spnGrp_SCH_ID|spnGrp_pay_id |
+---+------+-------------+--------------+-----------+------------+-------------+--------------+
|  1|credit|          5.0|         105.0|        4.0|       104.0|          6.0|         106.0|
|  1| pay  |          2.0|         102.0|        1.0|       101.0|          3.0|         103.0|
+---+------+-------------+--------------+-----------+------------+-------------+--------------+

第二次旋转之后

+---+--------------------+---------------------+------------------+-------------------+--------------------+---------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+
| ID|credit_client_sch_id|credit_client_pay_id | credit_phr_SCH_ID| credit_phr_pay_id |credit_spnGrp_SCH_ID|credit_spnGrp_pay_id |pay_client_sch_id|pay_client_pay_id |   pay_phr_SCH_ID|   pay_phr_pay_id |pay_spnGrp_SCH_ID|pay_spnGrp_pay_id |
+---+--------------------+---------------------+------------------+-------------------+--------------------+---------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+
|  1|                 5.0|                105.0|               4.0|              104.0|                 6.0|                106.0|             2.0|              102.0|              1.0|             101.0|              3.0|             103.0|
+---+--------------------+---------------------+------------------+-------------------+--------------------+---------------------+-----------------+------------------+-----------------+------------------+-----------------+------------------+

虽然我不确定性能。

cig3rfwq

cig3rfwq3#

df.createOrReplaceTempView("NumberDF")

df.createOrReplaceTempView("MasterDf")

val sqlDF = spark.sql("select m.id,t1.fld1,t2.fld2,t3.fld3,m.otherfields 
from 
(select id, (case when n.cat='pay' and n.sub_cat ='client' then n.sch_id end) fld1
from NumberDF n where case when n.cat='pay' and n.sub_cat ='client' then n.sch_id end is not null ) t1 ,
(select id, (case when  n.cat='pay' and n.sub_cat ='phr' then n.sch_id end) fld2
from NumberDF n where case when  n.cat='pay' and n.sub_cat ='phr' then n.sch_id end is not null ) t2,
(select id, (case when  n.cat='credit' and n.sub_cat ='spGrp' then n.pay_id end) fld3
from NumberDF n where case when  n.cat='credit' and n.sub_cat ='spGrp' then n.pay_id end is not null ) t3,
MasterDf m ")

sqlDF.show()

相关问题