extraire des données à partir de la table de la ruche dans spark et effectuer la jointure sur RDDs

J'ai deux tables dans la ruche/impala. Je veux extraire les données de la table dans spark comme rdd et dire effectuer une opération de jointure.

Je ne veux pas passer directement la requête de jointure dans ma ruche contexte. C'est juste un exemple.
J'ai plus de cas d'utilisation qui ne sont pas possible par une norme HiveQL.
Comment puis-je récupérer toutes les lignes, l'accès aux colonnes et effectuer la transformation.

Supposons que j'ai deux rdd:

val table1 =  hiveContext.hql("select * from tem1")

val table2 =  hiveContext.hql("select * from tem2")

Je veux faire une jointure sur la rdd sur une colonne appelée "account_id"

Idéalement, je veux faire quelque chose comme ceci en utilisant la rdd à l'aide de l'étincelle shell.

select * from tem1 join tem2 on tem1.account_id=tem2.account_id;

OriginalL'auteur user1189851 | 2014-11-06

  1. 2

    Je ne suis pas sûr que j'ai compris la question, mais comme une alternative, vous pouvez utiliser l'API pour rejoindre DataFrames, de sorte que vous pouvez avoir beaucoup de choses décidé par programmation (par exemple, la join fonction peut être passée en paramètre à une méthode qui s'applique une transformation personnalisée).

    Pour votre exemple, il serait comme ceci:

    val table1 =  hiveContext.sql("select * from tem1")
val table2 =  hiveContext.sql("select * from tem2")
val common_attributes = Seq("account_id")
val joined = table1.join(table2, common_attributes)

    Il y a de nombreuses transformations disponibles dans le DataFrame de l'API:
    http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrame

    Acclamations

    OriginalL'auteur Daniel de Paula

  2. 1

    Donc, nous pourrions inscrire table1 et table2 que les tables temporaires et ensuite faire la jointure sur ces tables temporaires.

    table1.registerTempTable("t1")
table2.registerTempTable("t2")
table3 = hiveContext.hql("select * from t1 join t2 on t1.account_id=t2.account_id")
    Salut holden, merci, mais je l'ai déjà mentionné que je ne veux pas faire de cette façon. C'est juste un exemple simple. J'ai un cas d'utilisation où j'ai des requêtes plus complexes. Je veux être capable de construire des rdd est à partir d'un ensemble de résultats et d'effectuer la jointure et d'autres opérations.
    Ah désolé user1189851, je pensais que tu voulais éviter de faire une jointure sur l'origine de la ruche tables. Dans le code ci-dessus l'extrait de code que j'ai posté table1 et table2 peut être tout SchemaRDD (et notez que les requêtes que nous écrivons nous redonner SchemaRDDs), que vouliez-vous table1 et table2? Voulez-vous être à partir du non-Spark SQL sources?
    j'ai donc un cas où mon premier rdd est un jeu de résultats d'une requête. et le second est un jeu de résultats d'une autre requête. val rdd1 = hiveContext.hql("select * from table1") et val rdd2. = hiveContext.hql("select * from table2"). Je veux dire effectuer la jointure sur ces deux rdds sur un attribut commun appelé account_id. L'idée est que je ne veux pas la rejoindre à l'intérieur de la ruche contexte, je devrais être capable de le faire en utilisant les transformations.

    OriginalL'auteur Holden

  3. 0

    table1 et table2 sont de type DataFrame. Il est possible de les transformer en ca de l'aide:

    lazy val table1_rdd = table1.rdd
lazy val table2_rdd = table2.rdd

    Cela devrait le truc. Sur ces rdd vous pouvez utiliser n'importe quel rdd opération.

    Voir aussi: https://issues.apache.org/jira/browse/SPARK-6608 et https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrame

    OriginalL'auteur Blaubaer

  4. 0

    Vous pouvez directement sélectionner la colonne dont vous voulez de code suivant:

    val table1 =  hiveContext.hql("select account_id from tem1")
val table2 =  hiveContext.hql("select account_id from tem2")
val joinedTable = table1.join(table2)

    OriginalL'auteur BadBoy777