Splitting Saiten auf zwei verschiedene Trennzeichen

Question

Ich habe eine Reihe von Daten, die etwa wie folgt aussieht:

1:a:x|y|z 
2:b:y|z 
3:c:x 
4:d:w|x 

Was ich will, ist ein Ausgang, der wie folgt aussieht:

1,a,x 
1,a,y 
1,a,z 
2,b,y 
2,b,z 
3,c,x 
4,d,w 
4,d,x 

Ich habe versucht, Aufteilen sowohl auf ':' als auch auf '|' aber es hat nicht geholfen, da es ein Ergebnis wie dieses gab:

1,a,x,y,z 
2,b,y,z 
3,c,x 
4,d,w,x 

Auch ist es eine Möglichkeit, die unerwünschten Werte aus dem rdd herausfiltern, die ich bekomme, wenn ich so herauszufiltern (w, y, z) von

1,a,x,y,z 
2,b,y,z 
3,c,x 
4,d,w,x 

Erwartete Ausgabe aussehen wird:

1,a,x 
2,b,  //it'll be fine if this doesn't even appear, better in fact 
3,c,x 
4,d,x 

Irgendwelche Ideen?

Answer 1

ich die einzige Spalte gehe davon aus, dass mehrere Optionen enthalten kann, ist die letzte

rdd.flatMap { 
    row => 
    val Array(col1, col2, col3) = row.split(':') 
    col3.split('|').map(value => (col1, col2, value)) 
} 

Danach erhalten Sie eine RDD[(String, String, String)]

Answer 2

val df = sc.parallelize(Seq(("1:a:x|y|z"), ("2:b:y|z"), ("3:c:x"),("4:d:w|x"))) 
df.collect 
//flat map 
val df1=df.flatMap { 
    row => 
    val Array(col1, col2, col3) = row.split(':') 
    col3.split('|').map(value => (col1, col2, value)) 
} 

df1.collect 
//filer as per requirments 
val df2=df1.toDF("col1","col2","col3") 
df2.show() 
//df2.createOrReplaceTempView("TempTable") 
//val countDF = spark.sqlContext.sql("SELECT col1,col2,col3, MIN(col1) FROM TempTable GROUP BY col1,col2,col3").show() 

val w = Window.partitionBy($"col1").orderBy($"col1".desc) 

val dfTop = df2.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn").orderBy($"col1".asc) 

dfTop.show 

Ergebnisse erhalten:

+----+----+----+ 
|col1|col2|col3| 
+----+----+----+ 
| 1| a| x| 
| 1| a| y| 
| 1| a| z| 
| 2| b| y| 
| 2| b| z| 
| 3| c| x| 
| 4| d| w| 
| 4| d| x| 
+----+----+----+ 

Nach Filter

+----+----+----+ 
|col1|col2|col3| 
+----+----+----+ 
| 1| a| x| 
| 2| b| y| 
| 3| c| x| 
| 4| d| w| 
+----+----+----+ 

df: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[88] at parallelize at command-4102047427741428:3 
df1: org.apache.spark.rdd.RDD[(String, String, String)] = MapPartitionsRDD[89] at flatMap at command-4102047427741428:6 
df2: org.apache.spark.sql.DataFrame = [col1: string, col2: string ... 1 more field] 
w: org.apache.spark.sql.expressions.WindowSpec = org.apache.spark.sql.expressions.WindowSpec@31cd5345 
dfTop: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [col1: string, col2: string ... 1 more field] 
Command took 1.01 seconds -- by vaquar.khan@gmail.com at 11/11/2017, 10:35:31 PM on My Cluster 

Answer 3

Erster Schritt wäre die Aufteilung für jede Zeile sein:

"1:a:x|y|z".split(':').toList.map(_.split('|').toList) 

---- 
in: List[List[String]] = List(List(1), List(a), List(x, y, z)) 

Dann erzeugen, die Kombinationen:

def combine(input: List[List[String]]): List[List[String]] = input match { 
    case x :: xs => x.flatMap(s => combine(xs).map(s :: _)) 
    case Nil  => List(Nil) 
    } 

    val res = combine(in).map(_.mkString(",")) 

    ---- 
    res: List[String] = List(1,a,x, 1,a,y, 1,a,z) 

Dann können Sie .filter bei Bedarf:

res.filter(_.contains("x"))