Concaténer les colonnes dans Apache Spark DataFrame

Comment concaténer deux colonnes dans un DataFrame Apache Spark? Existe-t-il une fonction dans Spark SQL que nous pouvons utiliser?

Avec le SQL brut, vous pouvez utiliser CONCAT:

• En Python

  df = sqlContext.createDataFrame([("foo", 1), ("bar", 2)], ("k", "v"))
  df.registerTempTable("df")
  sqlContext.sql("SELECT CONCAT(k, ' ',  v) FROM df")
  

• Dans Scala

  import sqlContext.implicits._
  
  val df = sc.parallelize(Seq(("foo", 1), ("bar", 2))).toDF("k", "v")
  df.registerTempTable("df")
  sqlContext.sql("SELECT CONCAT(k, ' ',  v) FROM df")
  

Depuis Spark 1.5.0, vous pouvez utiliser la concatfonction avec l'API DataFrame:

• En Python:

  from pyspark.sql.functions import concat, col, lit
  
  df.select(concat(col("k"), lit(" "), col("v")))
  

• Dans Scala:

  import org.apache.spark.sql.functions.{concat, lit}
  
  df.select(concat($"k", lit(" "), $"v"))
  

Il existe également une concat_wsfonction qui prend un séparateur de chaîne comme premier argument.

Voici comment vous pouvez attribuer un nom personnalisé

import pyspark
from pyspark.sql import functions as sf
sc = pyspark.SparkContext()
sqlc = pyspark.SQLContext(sc)
df = sqlc.createDataFrame([('row11','row12'), ('row21','row22')], ['colname1', 'colname2'])
df.show()

donne,

+--------+--------+
|colname1|colname2|
+--------+--------+
|   row11|   row12|
|   row21|   row22|
+--------+--------+

créer une nouvelle colonne en concaténant:

df = df.withColumn('joined_column', 
                    sf.concat(sf.col('colname1'),sf.lit('_'), sf.col('colname2')))
df.show()

+--------+--------+-------------+
|colname1|colname2|joined_column|
+--------+--------+-------------+
|   row11|   row12|  row11_row12|
|   row21|   row22|  row21_row22|
+--------+--------+-------------+

Une option pour concaténer les colonnes de chaînes dans Spark Scala utilise concat.

Il est nécessaire de vérifier les valeurs nulles . Parce que si l'une des colonnes est nulle, le résultat sera nul même si l'une des autres colonnes contient des informations.

Utilisation concatet withColumn:

val newDf =
  df.withColumn(
    "NEW_COLUMN",
    concat(
      when(col("COL1").isNotNull, col("COL1")).otherwise(lit("null")),
      when(col("COL2").isNotNull, col("COL2")).otherwise(lit("null"))))

Utilisation concatet select:

val newDf = df.selectExpr("concat(nvl(COL1, ''), nvl(COL2, '')) as NEW_COLUMN")

Avec les deux approches, vous aurez un NEW_COLUMN dont la valeur est une concaténation des colonnes: COL1 et COL2 de votre df d'origine.

Si vous souhaitez le faire en utilisant DF, vous pouvez utiliser un udf pour ajouter une nouvelle colonne basée sur des colonnes existantes.

val sqlContext = new SQLContext(sc)
case class MyDf(col1: String, col2: String)

//here is our dataframe
val df = sqlContext.createDataFrame(sc.parallelize(
    Array(MyDf("A", "B"), MyDf("C", "D"), MyDf("E", "F"))
))

//Define a udf to concatenate two passed in string values
val getConcatenated = udf( (first: String, second: String) => { first + " " + second } )

//use withColumn method to add a new column called newColName
df.withColumn("newColName", getConcatenated($"col1", $"col2")).select("newColName", "col1", "col2").show()

Depuis Spark 2.3 ( SPARK-22771 ) Spark SQL prend en charge l'opérateur de concaténation ||.

Par exemple;

val df = spark.sql("select _c1 || _c2 as concat_column from <table_name>")

Voici une autre façon de faire cela pour pyspark:

#import concat and lit functions from pyspark.sql.functions 
from pyspark.sql.functions import concat, lit

#Create your data frame
countryDF = sqlContext.createDataFrame([('Ethiopia',), ('Kenya',), ('Uganda',), ('Rwanda',)], ['East Africa'])

#Use select, concat, and lit functions to do the concatenation
personDF = countryDF.select(concat(countryDF['East Africa'], lit('n')).alias('East African'))

#Show the new data frame
personDF.show()

----------RESULT-------------------------

84
+------------+
|East African|
+------------+
|   Ethiopian|
|      Kenyan|
|     Ugandan|
|     Rwandan|
+------------+

Voici une suggestion lorsque vous ne connaissez pas le nombre ou le nom des colonnes dans le Dataframe.

val dfResults = dfSource.select(concat_ws(",",dfSource.columns.map(c => col(c)): _*))

concat (* cols)

v1.5 et supérieur

Concatène plusieurs colonnes d'entrée en une seule colonne. La fonction fonctionne avec des chaînes, des colonnes binaires et des tableaux compatibles.

Par exemple: new_df = df.select(concat(df.a, df.b, df.c))

concat_ws (sep, * cols)

v1.5 et supérieur

Similaire à concatmais utilise le séparateur spécifié.

Par exemple: new_df = df.select(concat_ws('-', df.col1, df.col2))

map_concat (* cols)

v2.4 et supérieur

Utilisé pour concater des cartes, renvoie l'union de toutes les cartes données.

Par exemple: new_df = df.select(map_concat("map1", "map2"))

Utilisation de l'opérateur de chaîne concat ( ||):

v2.3 et supérieur

Par exemple: df = spark.sql("select col_a || col_b || col_c as abc from table_x")

Référence: Spark sql doc

Dans Spark 2.3.0, vous pouvez faire:

spark.sql( """ select '1' || column_a from table_a """)

En Java, vous pouvez le faire pour concaténer plusieurs colonnes. L'exemple de code est de vous fournir un scénario et comment l'utiliser pour une meilleure compréhension.

SparkSession spark = JavaSparkSessionSingleton.getInstance(rdd.context().getConf());
Dataset<Row> reducedInventory = spark.sql("select * from table_name")
                        .withColumn("concatenatedCol",
                                concat(col("col1"), lit("_"), col("col2"), lit("_"), col("col3")));


class JavaSparkSessionSingleton {
    private static transient SparkSession instance = null;

    public static SparkSession getInstance(SparkConf sparkConf) {
        if (instance == null) {
            instance = SparkSession.builder().config(sparkConf)
                    .getOrCreate();
        }
        return instance;
    }
}

Le code ci-dessus concaténé col1, col2, col3 séparés par "_" pour créer une colonne avec le nom "concatenatedCol".

Avons-nous la syntaxe Java correspondant au processus ci-dessous

val dfResults = dfSource.select(concat_ws(",",dfSource.columns.map(c => col(c)): _*))

Une autre façon de le faire dans pySpark en utilisant sqlContext ...

#Suppose we have a dataframe:
df = sqlContext.createDataFrame([('row1_1','row1_2')], ['colname1', 'colname2'])

# Now we can concatenate columns and assign the new column a name 
df = df.select(concat(df.colname1, df.colname2).alias('joined_colname'))

En effet, il existe de belles abstractions intégrées pour que vous puissiez accomplir votre concaténation sans avoir besoin d'implémenter une fonction personnalisée. Depuis que vous avez mentionné Spark SQL, je suppose que vous essayez de le passer en tant que commande déclarative via spark.sql (). Si tel est le cas, vous pouvez effectuer de manière simple en passant une commande SQL comme: SELECT CONCAT(col1, '<delimiter>', col2, ...) AS concat_column_name FROM <table_name>;

De plus, à partir de Spark 2.3.0, vous pouvez utiliser des commandes en lignes avec: SELECT col1 || col2 AS concat_column_name FROM <table_name>;

Où, est votre délimiteur préféré (peut également être un espace vide) et est la table temporaire ou permanente que vous essayez de lire.

Nous pouvons également utiliser simplement SelectExpr. df1.selectExpr ("*", "supérieur (_2 || _3) comme nouveau")