Spark 2.2.0 API: Welches sollte ich Dataset mit Groupby kombiniert mit Aggregat oder RDD mit ReduceBykey bevorzugen

Question

Hallo alle, zunächst mit dem Titel auf der Basis kann jemand sagen, dass die Frage bereits beantwortet ist, aber mein Punkt ist zu vergleichen ReduceBykey, GroupBykey-Leistung, spezifisch für das Dataset und die RDD-API. Ich habe in vielen Beiträgen gesehen, dass die Leistung über die ReduceBykey-Methode gegenüber GroupByKey effizienter ist und natürlich stimme ich damit überein. Trotzdem bin ich etwas verwirrt und ich kann nicht herausfinden, wie sich diese Methoden verhalten, wenn wir ein Dataset oder RDD verwenden. Welcher sollte man jeweils verwenden?

Ich werde versuchen, genauer zu sein, so werde ich mein Problem mit meinem lösen sowie mit dem Arbeitscode zur Verfügung stellen und ich warte auf Ihre früheste Bequemlichkeit, um mir eine Verbesserungen zu diesem vorschlagen.

+---+------------------+-----+ 
|id |Text1    |Text2| 
+---+------------------+-----+ 
|1 |one,two,three  |one | 
|2 |four,one,five  |six | 
|3 |seven,nine,one,two|eight| 
|4 |two,three,five |five | 
|5 |six,five,one  |seven| 
+---+------------------+-----+ 

Der Punkt hier ist zu prüfen, ob die dritte Colum auf jeder Zeile der zweiten Colum enthalten ist, und danach, alle die ID thems sammeln. Zum Beispiel erschien das Wort der dritten Spalte «Eins» in den Sätzen der zweiten Spalte mit der ID 1, 5, 2, 3.

+-----+------------+ 
|Text2|Set   | 
+-----+------------+ 
|seven|[3]   | 
|one |[1, 5, 2, 3]| 
|six |[5]   | 
|five |[5, 2, 4] | 
+-----+------------+ 

Hier ist mein Arbeits Code

List<Row> data = Arrays.asList(
       RowFactory.create(1, "one,two,three", "one"), 
       RowFactory.create(2, "four,one,five", "six"), 
       RowFactory.create(3, "seven,nine,one,two", "eight"), 
       RowFactory.create(4, "two,three,five", "five"), 
       RowFactory.create(5, "six,five,one", "seven") 
     ); 

     StructType schema = new StructType(new StructField[]{ 
       new StructField("id", DataTypes.IntegerType, false, Metadata.empty()), 
       new StructField("Text1", DataTypes.StringType, false, Metadata.empty()), 
       new StructField("Text2", DataTypes.StringType, false, Metadata.empty()) 
     }); 

     Dataset<Row> df = spark.createDataFrame(data, schema); 
     df.show(false); 
     Dataset<Row> df1 = df.select("id", "Text1") 
       .crossJoin(df.select("Text2")) 
       .filter(col("Text1").contains(col("Text2"))) 
       .orderBy(col("Text2")); 

     df1.show(false); 

     Dataset<Row> df2 = df1 
       .groupBy("Text2") 
       .agg(collect_set(col("id")).as("Set")); 

     df2.show(false); 

Meine Frage ist, Detail in 3 Teilfolgen:

• um tun, um die Leistung zu verbessern und ich den Datensatz in RDD konvertieren müssen und ReduceBykey statt Dataset GROUPBY machen?
• Welches sollte ich verwenden und warum? Dataset oder RDD
• ich wäre dankbar, wenn Sie eine alternative Lösung geben könnte, die effizienter, wenn existiert in meinem Ansatz ist

Answer 1

TL; DR Beide sind schlecht, aber wenn Sie mit Dataset Aufenthalt mit Dataset.

Dataset.groupBy verhält sich wie reduceByKey bei Verwendung mit geeigneter Funktion. Leider verhält sich collect_set ähnlich wie groupByKey, wenn die Anzahl der Duplikate gering ist. Umschreiben mit reduceByKeywon't change a thing.

würde ich Ihnen dankbar, wenn Sie eine alternative Lösung geben könnte, die effizienter ist, wenn in meinem Ansatz besteht

besten Sie tun können, entfernen crossJoin:

val df = Seq((1, "one,two,three", "one"), 
    (2, "four,one,five", "six"), 
    (3, "seven,nine,one,two", "eight"), 
    (4, "two,three,five", "five"), 
    (5, "six,five,one", "seven")).toDF("id", "text1", "text2") 

df.select(col("id"), explode(split(col("Text1"), ",")).alias("w")) 
    .join(df.select(col("Text2").alias("w")), Seq("w")) 
    .groupBy("w") 
    .agg(collect_set(col("id")).as("Set")).show 

+-----+------------+ 
| w|   Set| 
+-----+------------+ 
|seven|   [3]| 
| one|[1, 5, 2, 3]| 
| six|   [5]| 
| five| [5, 2, 4]| 
+-----+------------+