Joindre deux ordinaire RDDs avec/sans Spark SQL

J'ai besoin de joindre deux ordinaire RDDs sur une ou plusieurs colonnes. Logiquement, cette opération est équivalente à la base de données opération de jointure de deux tables. Je me demande si ce n'est possible que par Spark SQL ou il y a d'autres façons de le faire.

Comme un exemple concret, pensez
CA r1 avec une clé primaire ITEM_ID:

(ITEM_ID, ITEM_NAME, ITEM_UNIT, COMPANY_ID)

et CA r2 avec une clé primaire COMPANY_ID:

(COMPANY_ID, COMPANY_NAME, COMPANY_CITY)

Je veux rejoindre r1 et r2.

Comment cela peut-il être fait?

InformationsquelleAutor learning_spark | 2014-12-12
  1. 25

    Soumya Simanta a donné une bonne réponse. Cependant, les valeurs de a rejoint RDD sont Iterable, de sorte que les résultats peuvent ne pas être très similaire à table ordinaire de rejoindre.

    Alternativement, vous pouvez:

    val mappedItems = items.map(item => (item.companyId, item))
val mappedComp = companies.map(comp => (comp.companyId, comp))
mappedItems.join(mappedComp).take(10).foreach(println)

    La sortie serait:

    (c1,(Item(1,first,2,c1),Company(c1,company-1,city-1)))
(c1,(Item(2,second,2,c1),Company(c1,company-1,city-1)))
(c2,(Item(3,third,2,c2),Company(c2,company-2,city-2)))
    • Vos cartes sont les mêmes que items.keyBy{_.companyId}, companies.keyBy{_.companyId}. Depuis qu'ils font partie d'Étincelle, il y a une chance que ce serait plus efficace?
    • C'est l'étincelle de code source pour keyBy: def keyBy[K](f: T => K): RDD[(K, T)] = { map(x => (f(x), x)) } donc votre solution et @virya solution sont tout à fait la même
    • OK 🙂 Encore, peut-être l'intention légèrement plus claire avec keyBy. Pas un point important, si
    InformationsquelleAutor viirya
  2. 12

    (À L'Aide De La Scala)
    Disons que vous avez deux Rdd:

    • emp: (empid, ename, dept)

    • dept: (dname, dept)

    - Ci est une autre façon:

    //val emp = sc.parallelize(Seq((1,"jordan",10), (2,"ricky",20), (3,"matt",30), (4,"mince",35), (5,"rhonda",30)))
val emp = sc.parallelize(Seq(("jordan",10), ("ricky",20), ("matt",30), ("mince",35), ("rhonda",30)))
val dept = sc.parallelize(Seq(("hadoop",10), ("spark",20), ("hive",30), ("sqoop",40)))
//val shifted_fields_emp = emp.map(t => (t._3, t._1, t._2))
val shifted_fields_emp = emp.map(t => (t._2, t._1))
val shifted_fields_dept = dept.map(t => (t._2,t._1))
shifted_fields_emp.join(shifted_fields_dept)
//Create emp RDD
val emp = sc.parallelize(Seq((1,"jordan",10), (2,"ricky",20), (3,"matt",30), (4,"mince",35), (5,"rhonda",30)))
//Create dept RDD
val dept = sc.parallelize(Seq(("hadoop",10), ("spark",20), ("hive",30), ("sqoop",40)))
//Establishing that the third field is to be considered as the Key for the emp RDD
val manipulated_emp = emp.keyBy(t => t._3)
//Establishing that the second field need to be considered as the Key for dept RDD
val manipulated_dept = dept.keyBy(t => t._2)
//Inner Join
val join_data = manipulated_emp.join(manipulated_dept)
//Left Outer Join
val left_outer_join_data = manipulated_emp.leftOuterJoin(manipulated_dept)
//Right Outer Join
val right_outer_join_data = manipulated_emp.rightOuterJoin(manipulated_dept)
//Full Outer Join
val full_outer_join_data = manipulated_emp.fullOuterJoin(manipulated_dept)
//Formatting the Joined Data for better understandable (using map)
val cleaned_joined_data = join_data.map(t => (t._2._1._1, t._2._1._2, t._1, t._2._2._1))

    Cela vous donnera le résultat sous la forme:

    //Imprimer la sortie cleaned_joined_data sur la console

    scala> cleaned_joined_data.collect()
res13: Array[(Int, String, Int, String)] = Array((3,matt,30,hive), (5,rhonda,30,hive), (2,ricky,20,spark), (1,jordan,10,hadoop))
    InformationsquelleAutor New Coder
  3. 9

    Quelque chose comme cela devrait fonctionner. 

    scala> case class Item(id:String, name:String, unit:Int, companyId:String)
scala> case class Company(companyId:String, name:String, city:String)
scala> val i1 = Item("1", "first", 2, "c1")
scala> val i2 = i1.copy(id="2", name="second")
scala> val i3 = i1.copy(id="3", name="third", companyId="c2")
scala> val items = sc.parallelize(List(i1,i2,i3))
items: org.apache.spark.rdd.RDD[Item] = ParallelCollectionRDD[14] at parallelize at <console>:20
scala> val c1 = Company("c1", "company-1", "city-1")
scala> val c2 = Company("c2", "company-2", "city-2")
scala> val companies = sc.parallelize(List(c1,c2))
scala> val groupedItems = items.groupBy( x => x.companyId) 
groupedItems: org.apache.spark.rdd.RDD[(String, Iterable[Item])] = ShuffledRDD[16] at groupBy at <console>:22
scala> val groupedComp = companies.groupBy(x => x.companyId)
groupedComp: org.apache.spark.rdd.RDD[(String, Iterable[Company])] = ShuffledRDD[18] at groupBy at <console>:20
scala> groupedItems.join(groupedComp).take(10).foreach(println)
14/12/12 00:52:32 INFO DAGScheduler: Job 5 finished: take at <console>:35, took 0.021870 s
(c1,(CompactBuffer(Item(1,first,2,c1), Item(2,second,2,c1)),CompactBuffer(Company(c1,company-1,city-1))))
(c2,(CompactBuffer(Item(3,third,2,c2)),CompactBuffer(Company(c2,company-2,city-2))))
    InformationsquelleAutor Soumya Simanta
  4. 2

    Spark SQL pouvez effectuer rejoindre sur la SPARK RDDs.

    Code ci-dessous effectue une jointure SQL sur l'Entreprise et les Éléments de Rdd

    object SparkSQLJoin {
case class Item(id:String, name:String, unit:Int, companyId:String)
case class Company(companyId:String, name:String, city:String)
def main(args: Array[String]) {
val sparkConf = new SparkConf()
val sc= new SparkContext(sparkConf)
val sqlContext = new SQLContext(sc)
import sqlContext.createSchemaRDD
val i1 = Item("1", "first", 1, "c1")
val i2 = Item("2", "second", 2, "c2")
val i3 = Item("3", "third", 3, "c3")
val c1 = Company("c1", "company-1", "city-1")
val c2 = Company("c2", "company-2", "city-2")
val companies = sc.parallelize(List(c1,c2))
companies.registerAsTable("companies")
val items = sc.parallelize(List(i1,i2,i3))
items.registerAsTable("items")
val result = sqlContext.sql("SELECT * FROM companies C JOIN items I ON C.companyId= I.companyId").collect
result.foreach(println)
}
}

    Sortie est affichée comme 

         [c1,company-1,city-1,1,first,1,c1]
[c2,company-2,city-2,2,second,2,c2]
    • J'ai plusieurs colonnes, donc j'ai besoin de spécifier le schéma de la programmation. Aussi les Rdd sont créés à partir de grands fichiers de texte sur HDFS. Je crois que l'approche ci-dessus fonctionne toujours, droite? S'il vous plaît laissez-moi savoir si des changements sont nécessaires.
    • Oui, cette approche fonctionne bien pour les données énorme aussi. Pour la définition du schéma par programme découvrez spark.apache.org/docs/latest/...
    InformationsquelleAutor Vijay Innamuri