Mam następujący kod w SCALA:

val  fullCertificateSourceDf = certificateSourceDf         
              .withColumn("Stage", when(col("Data.WorkBreakdownUp1Summary").isNotNull && col("Data.WorkBreakdownUp1Summary")=!="",                                                     rtrim(regexp_extract($"Data.WorkBreakdownUp1Summary","^.*?(?= - *[a-zA-Z])",0))).otherwise(""))
              .withColumn("SubSystem", when(col("Data.ProcessBreakdownSummaryList").isNotNull && col("Data.ProcessBreakdownSummaryList")=!="",                                         regexp_extract($"Data.ProcessBreakdownSummaryList","^.*?(?= - *[a-zA-Z])",0)).otherwise(""))
              .withColumn("System", when(col("Data.ProcessBreakdownUp1SummaryList").isNotNull && col("Data.ProcessBreakdownUp1SummaryList")=!="",                                         regexp_extract($"Data.ProcessBreakdownUp1SummaryList","^.*?(?= - *[a-zA-Z])",0)).otherwise(""))
              .withColumn("Facility", when(col("Data.ProcessBreakdownUp2Summary").isNotNull && col("Data.ProcessBreakdownUp2Summary")=!="",                                         regexp_extract($"Data.ProcessBreakdownUp2Summary","^.*?(?= - *[a-zA-Z])",0)).otherwise(""))
              .withColumn("Area", when(col("Data.ProcessBreakdownUp3Summary").isNotNull && col("Data.ProcessBreakdownUp3Summary")=!="",                                         regexp_extract($"Data.ProcessBreakdownUp3Summary","^.*?(?= - *[a-zA-Z])",0)).otherwise(""))
              .select("Data.ID",
                      "Data.CertificateID",
                      "Data.CertificateTag",
                      "Data.CertificateDescription",
                      "Data.WorkBreakdownUp1Summary",
                      "Data.ProcessBreakdownSummaryList",
                      "Data.ProcessBreakdownUp1SummaryList",
                      "Data.ProcessBreakdownUp2Summary",
                      "Data.ProcessBreakdownUp3Summary",
                      "Data.ActualStartDate",
                      "Data.ActualEndDate",
                      "Data.ApprovedDate",
                      "Data.CurrentState",
                      "DataType",
                      "PullDate",
                      "PullTime",
                      "Stage",
                      "System",
                      "SubSystem",
                      "Facility",
                      "Area"
                     )
                     .filter((col("Stage").isNotNull) && (length(col("Stage"))>0))
                     .filter(((col("SubSystem").isNotNull) && (length(col("SubSystem"))>0)) || ((col("System").isNotNull) && (length(col("System"))>0)) || ((col("Facility").isNotNull) && (length(col("Facility"))>0)) || ((col("Area").isNotNull) && (length(col("Area"))>0))
                      )
                     .select("*")

Ten dataframe FullCertificatesourctedF zawiera następujące dane:

Original Data

Ukryłem kilka kolumn na zwięzłość.

Chcę, żeby dane wyglądały tak:

Target Data

Dzielimy się na dwóch kolumnach: processbreakdownsummaryList i processbreakdownup1sumaryList. Oboje są wydzielonymi przecinkami listami.

Należy pamiętać, czy wartości w processiesbreakdownsummaryList (CS10-100-22-10 - system podgrzewacza wentylatora powietrza kopalniczego, CS10-100-81 -10 - Schnigularka kopalniowa) i processbreakdownup1SmaryList (CS10-100-22 - Wentylacja wału serwisowego, CS10-100-81 - Wał serwisowy Elektryczny) są takie same, że powinniśmy tylko dzielić się tylko raz.

Jednakże, jeśli są one różne, jak w processbreakdownsummaryList (CS10-100-22-10 - system podgrzewacza wentylatora powietrza kopalnianego, CS10-100-81 -10 - Schnigulatory Mine Schematgear) i processbreakdownUp1List (CS10-100) -22 - Wentylacja wału serwisowego, CS10-100-34 - Wał serwisowy Elektryczny) Należy podzielić ponownie przez trzeci rząd.

Z góry dziękuję za pomoc w tym pomocy.

0
user2227484 14 styczeń 2020, 22:49

1 odpowiedź

Najlepsza odpowiedź

Możesz rozwiązać to wiele sposobów, myślę, że najłatwiejszym podejściem do skomplikowanego przetwarzania jest użycie SCALA. Możesz przeczytać wszystkie kolumny, w tym "processbreakdownsummarylist" i "processbreakdownup1summarylist", porównaj ich wartości dla bycia samym / innym i emitują wiele wierszy dla jednego wiersza wejściowego. Następnie Flatmap na wyjściu, aby uzyskać dataframe ze wszystkimi potrzebnymi wierszami.

val fullCertificateSourceDf = // your code

fullCertificateSourceDf.map{ row =>
val id = row.getAs[String]("Data.ID")
... read all columns

val processBreakdownSummaryList = row.getAs[String]("Data.ProcessBreakdownSummaryList")
val processBreakdownUp1SummaryList = row.getAs[String]("Data.ProcessBreakdownUp1SummaryList")

//split processBreakdownSummaryList on ","
//split processBreakdownUp1SummaryList on ","
//compare then for equality 
//lets say you end up with 4 rows.

//return Seq of those 4 rows in a list processBreakdownSummary
//return a List of tuple of strings like List((id, certificateId, certificateTag, ..distinct values of processBreakdownUp1SummaryList...), (...) ...)
//all columns id, certificateId, certificateTag etc are repeated for each distinct value of processBreakdownUp1SummaryList and processBreakdownSummaryList

}.flatMap(identity(_)).toDF("column1","column2"...)

Oto przykład podziału jednego rzędu na wiele

    val employees = spark.createDataFrame(Seq(("E1",100.0,"a,b"), ("E2",200.0,"e,f"),("E3",300.0,"c,d"))).toDF("employee","salary","clubs")

    employees.map{ r =>
      val clubs = r.getAs[String]("clubs").split(",")
      for{
        c : String <- clubs
      }yield(r.getAs[String]("employee"),r.getAs[Double]("salary"), c)
    }.flatMap(identity(_)).toDF("employee","salary","clubs").show(false)

Wynik wygląda jak

+--------+------+-----+
|employee|salary|clubs|
+--------+------+-----+
|E1      |100.0 |a    |
|E1      |100.0 |b    |
|E2      |200.0 |e    |
|E2      |200.0 |f    |
|E3      |300.0 |c    |
|E3      |300.0 |d    |
+--------+------+-----+
0
Salim 15 styczeń 2020, 13:46