Comment effectuer des jointures de base de deux tables RDD dans Spark en utilisant Python?

Comment voulez-vous, la base se joint à Étincelle à l'aide de python? Dans R, vous pourriez utiliser merg() pour ce faire. Quelle est la syntaxe à l'aide de python sur la spark pour:

  1. Inner Join
  2. Jointure Externe Gauche
  3. Jointure Croisée

Avec deux tables (RDD) avec une seule colonne dans chaque qui a une clé commune.

RDD(1):(key,U)
RDD(2):(key,V)

Je pense qu'une jointure interne est quelque chose comme ceci:

rdd1.join(rdd2).map(case (key, u, v) => (key, ls ++ rs));

Est ce que le droit? J'ai cherché sur internet et ne pouvez pas trouver un bon exemple de jointures. Merci à l'avance.

source d'informationauteur invoketheshell

  1. 27

    Il peut être fait soit en utilisant PairRDDFunctions d'Étincelle ou de Trames de Données. Depuis trame de données des opérations de bénéficier de Catalyseur De L'Optimiseur la deuxième option est utile d'examiner.

    En supposant que vos données se présente comme suit: 

    rdd1 =  sc.parallelize([("foo", 1), ("bar", 2), ("baz", 3)])
rdd2 =  sc.parallelize([("foo", 4), ("bar", 5), ("bar", 6)])

    Avec PairRDDs:

    Inner join: 

    rdd1.join(rdd2)

    Jointure externe gauche:

    rdd1.leftOuterJoin(rdd2)

    Produit cartésien (ne nécessite pas de RDD[(T, U)]):

    rdd1.cartesian(rdd2)

    Diffusion de jointure (ne nécessite pas de RDD[(T, U)]):

    Enfin, il y a cogroup qui n'a pas de SQL directe équivalent, mais peut être utile dans certaines situations:

    cogrouped = rdd1.cogroup(rdd2)

cogrouped.mapValues(lambda x: (list(x[0]), list(x[1]))).collect()
## [('foo', ([1], [4])), ('bar', ([2], [5, 6])), ('baz', ([3], []))]

    Avec Spark Trames De Données

    Vous pouvez utiliser SQL DSL ou d'exécuter du SQL brut à l'aide sqlContext.sql.

    df1 = spark.createDataFrame(rdd1, ('k', 'v1'))
df2 = spark.createDataFrame(rdd2, ('k', 'v2'))

# Register temporary tables to be able to use sqlContext.sql
df1.createTempView('df1')
df2.createTempView('df2')

    Inner join:

    # inner is a default value so it could be omitted
df1.join(df2, df1.k == df2.k, how='inner') 
spark.sql('SELECT * FROM df1 JOIN df2 ON df1.k = df2.k')

    Jointure externe gauche:

    df1.join(df2, df1.k == df2.k, how='left_outer')
spark.sql('SELECT * FROM df1 LEFT OUTER JOIN df2 ON df1.k = df2.k')

    Cross join (explicite de jointure croisée ou les changements de configuration sont nécessaires dans Spark. 2.0 - spark.sql.crossJoin.activé pour Spark 2.x):

    df1.crossJoin(df2)
spark.sql('SELECT * FROM df1 CROSS JOIN df2')

    df1.join(df2)
sqlContext.sql('SELECT * FROM df JOIN df2')

    Depuis de 1,6 (1,5 en Scala) pour chacun de ces peut être combiné avec broadcast fonction:

    from pyspark.sql.functions import broadcast

df1.join(broadcast(df2), df1.k == df2.k)

    pour effectuer la diffusion de jointure. Voir aussi Pourquoi mon BroadcastHashJoin est plus lent que ShuffledHashJoin Spark