Połącz kolumny w Apache Spark DataFrame

Jak połączyć dwie kolumny w ramce danych Apache Spark? Czy w Spark SQL jest jakaś funkcja, której możemy użyć?

Z surowym SQL możesz użyć CONCAT:

• W Pythonie

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

• W 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")

Od Spark 1.5.0 możesz używać concatfunkcji z DataFrame API:

• W Pythonie:

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

• W Scali:

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

Istnieje również concat_wsfunkcja, która jako pierwszy argument przyjmuje separator ciągu.

Oto, jak możesz wykonać niestandardowe nazewnictwo

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()

daje,

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

utwórz nową kolumnę przez konkatenację:

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|
+--------+--------+-------------+

Jedną z opcji łączenia kolumn ciągów w Spark Scala jest użycie concat.

Konieczne jest sprawdzenie wartości null . Ponieważ jeśli jedna z kolumn ma wartość null, wynik będzie pusty, nawet jeśli jedna z pozostałych kolumn zawiera informacje.

Korzystanie concati 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"))))

Korzystanie concati select:

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

W przypadku obu podejść otrzymasz NEW_COLUMN, którego wartość jest konkatenacją kolumn: COL1 i COL2 z oryginalnego pliku df.

Jeśli chcesz to zrobić za pomocą DF, możesz użyć udf, aby dodać nową kolumnę na podstawie istniejących kolumn.

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()

Od Spark 2.3 ( SPARK-22771 ) Spark SQL obsługuje operatora konkatenacji ||.

Na przykład;

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

Oto inny sposób na zrobienie tego dla 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|
+------------+

Oto sugestia, jeśli nie znasz liczby lub nazwy kolumn w Dataframe.

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

concat (* cols)

wersja 1.5 i nowsza

Łączy wiele kolumn wejściowych w jedną kolumnę. Funkcja działa z łańcuchami, kolumnami binarnymi i zgodnymi tablicami.

Na przykład: new_df = df.select(concat(df.a, df.b, df.c))

concat_ws (wrz, * cols)

wersja 1.5 i nowsza

Podobny do, concatale używa określonego separatora.

Na przykład: new_df = df.select(concat_ws('-', df.col1, df.col2))

map_concat (* cols)

v2.4 i nowsze

Służy do łączenia map, zwraca sumę wszystkich podanych map.

Na przykład: new_df = df.select(map_concat("map1", "map2"))

Korzystanie z operatora łączenia ciągów ( ||):

v2.3 i nowsze

Na przykład: df = spark.sql("select col_a || col_b || col_c as abc from table_x")

Odniesienie: Spark sql doc

W Spark 2.3.0 możesz:

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

W Javie możesz to zrobić, aby połączyć wiele kolumn. Przykładowy kod ma na celu dostarczenie scenariusza i sposobu jego użycia w celu lepszego zrozumienia.

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;
    }
}

Powyższy kod łączył kolumnę z kolumną1, kolumną2, kolumną3 oddzieloną znakiem „_”, aby utworzyć kolumnę o nazwie „concatenatedCol”.

Czy mamy składnię javy odpowiadającą poniższemu procesowi

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

Inny sposób na zrobienie tego w pySpark przy użyciu 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'))

Rzeczywiście, istnieje kilka pięknych wbudowanych abstrakcji, dzięki którym możesz wykonać konkatenację bez konieczności implementowania funkcji niestandardowej. Ponieważ wspomniałeś o Spark SQL, zgaduję, że próbujesz przekazać to jako polecenie deklaratywne przez spark.sql (). Jeśli tak, możesz w prosty sposób przekazać polecenie SQL, takie jak: SELECT CONCAT(col1, '<delimiter>', col2, ...) AS concat_column_name FROM <table_name>;

Ponadto od Spark 2.3.0 możesz używać poleceń w wierszach z: SELECT col1 || col2 AS concat_column_name FROM <table_name>;

Gdzie jest twoim preferowanym separatorem (może to być również puste miejsce) i jest tymczasową lub stałą tabelą, z której próbujesz czytać.

Możemy również w prosty sposób użyć SelectExpr. df1.selectExpr ("*", "upper (_2 || _3) as new")