Pyspark - pivoter le dataframe

Question

J'ai ce dataframe en pyspark :

df = spark.createDataFrame([
("TenantId", "TennatId_1"),
("TimeGenerated", "2023-04-17T11:50:51.9013145Z"),
("ActivityType", "Connection"),
("CorrelationId", "608dd49a"),
("UserName", "test_1@test.cloud"),
("Name", "Name1"),
("Source", "Client"),
("Parameters", "{}"),
("SourceSystem", "Azure"),
("Type", "Check"),
("_ResourceId", "/subscriptions/5286ce"),
("TenantId", "TennatId_2"),
("TimeGenerated", "2023-04-17T11:50:51.944022Z"),
("ActivityType", "Connection"),
("CorrelationId", "11c0d75f0000"),
("UserName", "test_2@test.cloud"),
("Name", "Name2"),
("Source", "Client"),
("Parameters", "{}"),
("SourceSystem", "Azure"),
("Type", "Check"),
("_ResourceId", "/subscriptions/5286ce38-272f-4c54")], ["name", "rows"])

Et je veux faire un pivot dessus.

J'ai essayé l'expression ci-dessous :

pivoted_df = df.groupBy("name") \
           .pivot("name") \
           .agg(expr("first(rows) as rows")) \
           .orderBy("name")

mais j'obtiens cette sortie :

et mon désir est :

Comment cela peut-il être fait?

Answer 1

Utilisez les fonctions groupBy, Pivot sur le nom et agrégez sur les lignes pour obtenir le résultat souhaité.

Exemple:

from pyspark.sql.functions import *
df.groupBy(lit(1)).pivot("name").agg(first(col("rows"))).drop("1").show(10,False)
#+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------+
#|ActivityType|CorrelationId|Nom  |Paramètres|Source|SourceSystem|TenantId  |TimeGenerated               |Type |NomUtilisateur    |_Ressource           |
#+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------+
#|Connection  |608dd49a     |Name1|{}        |Client|Azure       |TennatId_1|2023-04-17T11:50:51.9013145Z|Check|test_1@test.cloud|/subscriptions/5286ce|
#+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------+

---

MISE À JOUR:

#define window
w=Window.partitionBy(lit("1")).orderBy("mid")

#ajouter une colonne d'identifiant d'ordre et une partition temporaire de fenêtre
df1 = df.withColumn("mid",monotonically_increasing_id()).\
  withColumn("temp_win", when(col("rows").rlike("^TennatId"),lit(1)).otherwise(lit(0))).\
  withColumn("windw", sum(col("temp_win")).over(w))

#pivot et fenêtre
df1.groupBy("windw").pivot("name").agg(first(col("rows"))).drop("windw").show(10,False)

#+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------------------+
|ActivityType|CorrelationId|Nom  |Paramètres|Source|SourceSystem|TenantId  |TimeGenerated               |Type |NomUtilisateur    |_Ressource                      |
+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------------------+
|Connection  |608dd49a     |Name1|{}        |Client|Azure       |TennatId_1|2023-04-17T11:50:51.9013145Z|Check|test_1@test.cloud|/subscriptions/5286ce            |
|Connection  |11c0d75f0000 |Name2|{}        |Client|Azure       |TennatId_2|2023-04-17T11:50:51.944022Z |Check|test_2@test.cloud|/subscriptions/5286ce38-272f-4c54|
#+------------+-------------+-----+----------+------+------------+----------+----------------------------+-----+-----------------+---------------------------------+