Spark SQL: Comment faire pour ajouter une nouvelle ligne à dataframe de table (à partir d'une autre table)

Je suis en utilisant Spark SQL avec dataframes. J'ai une entrée dataframe, et je tiens à ajouter (ou insert) ses lignes à un plus grand dataframe qui a plus de colonnes. Comment dois-je procéder?

Si c'était SQL, je voudrais utiliser INSERT INTO OUTPUT SELECT ... FROM INPUT, mais je ne sais pas comment le faire avec Spark SQL.

Pour le concret:

var input = sqlContext.createDataFrame(Seq(
(10L, "Joe Doe", 34),
(11L, "Jane Doe", 31),
(12L, "Alice Jones", 25)
)).toDF("id", "name", "age")
var output = sqlContext.createDataFrame(Seq(
(0L, "Jack Smith", 41, "yes", 1459204800L),
(1L, "Jane Jones", 22, "no", 1459294200L),
(2L, "Alice Smith", 31, "", 1459595700L)
)).toDF("id", "name", "age", "init", "ts")
scala> input.show()
+---+-----------+---+
| id|       name|age|
+---+-----------+---+
| 10|    Joe Doe| 34|
| 11|   Jane Doe| 31|
| 12|Alice Jones| 25|
+---+-----------+---+
scala> input.printSchema()
root
|-- id: long (nullable = false)
|-- name: string (nullable = true)
|-- age: integer (nullable = false)
scala> output.show()
+---+-----------+---+----+----------+
| id|       name|age|init|        ts|
+---+-----------+---+----+----------+
|  0| Jack Smith| 41| yes|1459204800|
|  1| Jane Jones| 22|  no|1459294200|
|  2|Alice Smith| 31|    |1459595700|
+---+-----------+---+----+----------+
scala> output.printSchema()
root
|-- id: long (nullable = false)
|-- name: string (nullable = true)
|-- age: integer (nullable = false)
|-- init: string (nullable = true)
|-- ts: long (nullable = false)

Je voudrais ajouter toutes les lignes de input à la fin de output. En même temps, je voudrais mettre l' output colonne de init être une chaîne vide '' et la ts colonne à le timestamp actuel, par exemple 1461883875L.

Toute aide serait appréciée.

Par la voie, pour la distribution de données il n'existe pas de "début" ou "fin", vous oubliez de l'index. En outre, en utilisant vars en Scala est connu comme une mauvaise pratique. Qu'est-ce que la valeur par défaut pour ts dans le schéma de vous montrer qu'il n'est pas nullable

OriginalL'auteur stackoverflowuser2010 | 2016-04-28

  1. 17

    Étincelle DataFrames sont immuables de sorte qu'il n'est pas possible d'ajouter /insérer des lignes. Au lieu de cela, vous pouvez simplement ajouter des colonnes manquantes et l'utilisation UNION ALL:

    output.unionAll(input.select($"*", lit(""), current_timestamp.cast("long")))
    C'est fantastique. Je vous remercie.
    Jolie, j'allais répondre à output.unionAll(input.select(input.columns.map(col) ++ List(lit("").as("init"), lit(0L).as("ts")) :_*)).show, mais votre réponse est évidemment mieux
    J'étais également en compte: var input2 = input.withColumn("init", lit(null:String)).withColumn("ts", current_timestamp.cast("long")) et ensuite de faire unionAll()

    OriginalL'auteur zero323

  2. 1

    J'ai eu un problème similaire correspondant à votre SQL-Question:

    Je voulais ajouter un dataframe à une table de la ruche, qui est aussi la plus grande (plus de colonnes). Pour garder votre exemple: output est ma table existante et input pourrait être le dataframe. Ma solution utilise simplement SQL et par souci d'exhaustivité, je tiens à fournir:

    import org.apache.spark.sql.SaveMode
var input = spark.createDataFrame(Seq(
(10L, "Joe Doe", 34),
(11L, "Jane Doe", 31),
(12L, "Alice Jones", 25)
)).toDF("id", "name", "age")
//--> just for a running example: In my case the table already exists
var output = spark.createDataFrame(Seq(
(0L, "Jack Smith", 41, "yes", 1459204800L),
(1L, "Jane Jones", 22, "no", 1459294200L),
(2L, "Alice Smith", 31, "", 1459595700L)
)).toDF("id", "name", "age", "init", "ts")
output.write.mode(SaveMode.Overwrite).saveAsTable("appendTest");
//<--
input.createOrReplaceTempView("inputTable");
spark.sql("INSERT INTO TABLE appendTest SELECT id, name, age, null, null FROM inputTable");
val df = spark.sql("SELECT * FROM appendTest")
df.show()

    sorties:

    +---+-----------+---+----+----------+
| id|       name|age|init|        ts|
+---+-----------+---+----+----------+
|  0| Jack Smith| 41| yes|1459204800|
|  1| Jane Jones| 22|  no|1459294200|
|  2|Alice Smith| 31|    |1459595700|
| 12|Alice Jones| 25|null|      null|
| 11|   Jane Doe| 31|null|      null|
| 10|    Joe Doe| 34|null|      null|
+---+-----------+---+----+----------+

    Si vous pouvez avoir ce problème, que vous ne savez pas comment autant de champs sont manquants, vous pouvez utiliser un diff comme

    val missingFields = output.schema.toSet.diff(input.schema.toSet)

    et ensuite (dans le mauvais pseudo-code)

    val sqlQuery = "INSERT INTO TABLE appendTest SELECT " + commaSeparatedColumnNames + commaSeparatedNullsForEachMissingField + " FROM inputTable"

    De l'espoir pour aider les personnes ayant des problèmes à l'avenir comme ça!

    P. S.: Dans votre cas particulier (timestamp actuel + champ vide pour init), vous pourriez même utiliser

    spark.sql("INSERT INTO TABLE appendTest SELECT id, name, age, '' as init, current_timestamp as ts FROM inputTable");

    qui résultats dans 

    +---+-----------+---+----+----------+
| id|       name|age|init|        ts|
+---+-----------+---+----+----------+
|  0| Jack Smith| 41| yes|1459204800|
|  1| Jane Jones| 22|  no|1459294200|
|  2|Alice Smith| 31|    |1459595700|
| 12|Alice Jones| 25|    |1521128513|
| 11|   Jane Doe| 31|    |1521128513|
| 10|    Joe Doe| 34|    |1521128513|
+---+-----------+---+----+----------+
    Est-il possible sans l'aide de SQL

    OriginalL'auteur Fabian