Jak dodać nową kolumnę do Spark DataFrame (przy użyciu PySpark)?

Mam Spark DataFrame (przy użyciu PySpark 1.5.1) i chciałbym dodać nową kolumnę.

Próbowałem następujących bez powodzenia:

type(randomed_hours) # => list

# Create in Python and transform to RDD

new_col = pd.DataFrame(randomed_hours, columns=['new_col'])

spark_new_col = sqlContext.createDataFrame(new_col)

my_df_spark.withColumn("hours", spark_new_col["new_col"])

Wystąpił również błąd podczas używania tego:

my_df_spark.withColumn("hours",  sc.parallelize(randomed_hours))

Jak więc dodać nową kolumnę (na podstawie wektora Python) do istniejącej ramki DataFrame za pomocą PySpark?

Nie możesz dodać dowolnej kolumny do DataFramew Spark. Nowe kolumny można tworzyć tylko przy użyciu literałów (inne typy literałów są opisane w Jak dodać stałą kolumnę w Spark DataFrame? )

from pyspark.sql.functions import lit

df = sqlContext.createDataFrame(
    [(1, "a", 23.0), (3, "B", -23.0)], ("x1", "x2", "x3"))

df_with_x4 = df.withColumn("x4", lit(0))
df_with_x4.show()

## +---+---+-----+---+
## | x1| x2|   x3| x4|
## +---+---+-----+---+
## |  1|  a| 23.0|  0|
## |  3|  B|-23.0|  0|
## +---+---+-----+---+

przekształcenie istniejącej kolumny:

from pyspark.sql.functions import exp

df_with_x5 = df_with_x4.withColumn("x5", exp("x3"))
df_with_x5.show()

## +---+---+-----+---+--------------------+
## | x1| x2|   x3| x4|                  x5|
## +---+---+-----+---+--------------------+
## |  1|  a| 23.0|  0| 9.744803446248903E9|
## |  3|  B|-23.0|  0|1.026187963170189...|
## +---+---+-----+---+--------------------+

zawarte za pomocą join:

from pyspark.sql.functions import exp

lookup = sqlContext.createDataFrame([(1, "foo"), (2, "bar")], ("k", "v"))
df_with_x6 = (df_with_x5
    .join(lookup, col("x1") == col("k"), "leftouter")
    .drop("k")
    .withColumnRenamed("v", "x6"))

## +---+---+-----+---+--------------------+----+
## | x1| x2|   x3| x4|                  x5|  x6|
## +---+---+-----+---+--------------------+----+
## |  1|  a| 23.0|  0| 9.744803446248903E9| foo|
## |  3|  B|-23.0|  0|1.026187963170189...|null|
## +---+---+-----+---+--------------------+----+

lub wygenerowane za pomocą funkcji / udf:

from pyspark.sql.functions import rand

df_with_x7 = df_with_x6.withColumn("x7", rand())
df_with_x7.show()

## +---+---+-----+---+--------------------+----+-------------------+
## | x1| x2|   x3| x4|                  x5|  x6|                 x7|
## +---+---+-----+---+--------------------+----+-------------------+
## |  1|  a| 23.0|  0| 9.744803446248903E9| foo|0.41930610446846617|
## |  3|  B|-23.0|  0|1.026187963170189...|null|0.37801881545497873|
## +---+---+-----+---+--------------------+----+-------------------+

Pod względem wydajności wbudowane funkcje ( pyspark.sql.functions), które są mapowane na wyrażenie Catalyst, są zwykle preferowane w stosunku do funkcji zdefiniowanych przez użytkownika w Pythonie.

Jeśli chcesz dodać zawartość dowolnego RDD jako kolumnę, możesz

• dodaj numery wierszy do istniejącej ramki danych
• wywołaj zipWithIndexRDD i przekonwertuj go na ramkę danych
• dołącz oba używając indeksu jako klucza łączenia

Aby dodać kolumnę za pomocą UDF:

df = sqlContext.createDataFrame(
    [(1, "a", 23.0), (3, "B", -23.0)], ("x1", "x2", "x3"))

from pyspark.sql.functions import udf
from pyspark.sql.types import *

def valueToCategory(value):
   if   value == 1: return 'cat1'
   elif value == 2: return 'cat2'
   ...
   else: return 'n/a'

# NOTE: it seems that calls to udf() must be after SparkContext() is called
udfValueToCategory = udf(valueToCategory, StringType())
df_with_cat = df.withColumn("category", udfValueToCategory("x1"))
df_with_cat.show()

## +---+---+-----+---------+
## | x1| x2|   x3| category|
## +---+---+-----+---------+
## |  1|  a| 23.0|     cat1|
## |  3|  B|-23.0|      n/a|
## +---+---+-----+---------+

Dla Spark 2.0

# assumes schema has 'age' column 
df.select('*', (df.age + 10).alias('agePlusTen'))

Istnieje wiele sposobów dodania nowej kolumny w pySpark.

Najpierw utwórzmy prostą ramkę DataFrame.

date = [27, 28, 29, None, 30, 31]
df = spark.createDataFrame(date, IntegerType())

Teraz spróbujmy podwoić wartość kolumny i zapisać ją w nowej kolumnie. PFB kilka różnych podejść do osiągnięcia tego samego.

# Approach - 1 : using withColumn function
df.withColumn("double", df.value * 2).show()

# Approach - 2 : using select with alias function.
df.select("*", (df.value * 2).alias("double")).show()

# Approach - 3 : using selectExpr function with as clause.
df.selectExpr("*", "value * 2 as double").show()

# Approach - 4 : Using as clause in SQL statement.
df.createTempView("temp")
spark.sql("select *, value * 2 as double from temp").show()

Więcej przykładów i wyjaśnień na temat funkcji Spark DataFrame można znaleźć na moim blogu .

Mam nadzieję, że to pomoże.

Możesz zdefiniować nowy udfpodczas dodawania column_name:

u_f = F.udf(lambda :yourstring,StringType())
a.select(u_f().alias('column_name')

from pyspark.sql.functions import udf
from pyspark.sql.types import *
func_name = udf(
    lambda val: val, # do sth to val
    StringType()
)
df.withColumn('new_col', func_name(df.old_col))

Chciałbym podać ogólny przykład dla bardzo podobnego przypadku użycia:

Przykład zastosowania: mam plik CSV składający się z:

First|Third|Fifth
data|data|data
data|data|data
...billion more lines

Muszę wykonać jakieś transformacje i końcowy plik csv musi wyglądać

First|Second|Third|Fourth|Fifth
data|null|data|null|data
data|null|data|null|data
...billion more lines

Muszę to zrobić, ponieważ jest to schemat zdefiniowany przez jakiś model i potrzebuję, aby moje końcowe dane były interoperacyjne z SQL Bulk Inserts i tym podobne.

więc:

1) Przeczytałem oryginalny plik csv za pomocą spark.read i nazywam go „df”.

2) Robię coś z danymi.

3) Dodaj puste kolumny za pomocą tego skryptu:

outcols = []
for column in MY_COLUMN_LIST:
    if column in df.columns:
        outcols.append(column)
    else:
        outcols.append(lit(None).cast(StringType()).alias('{0}'.format(column)))

df = df.select(outcols)

W ten sposób możesz ustrukturyzować swój schemat po załadowaniu pliku csv (będzie to również działać w przypadku zmiany kolejności kolumn, jeśli musisz to zrobić dla wielu tabel).

Najprostszym sposobem dodania kolumny jest użycie opcji „withColumn”. Ponieważ ramka danych jest tworzona za pomocą sqlContext, musisz określić schemat lub domyślnie może być dostępna w zestawie danych. Jeśli schemat jest określony, obciążenie staje się uciążliwe przy każdej zmianie.

Poniżej znajduje się przykład, który możesz wziąć pod uwagę:

from pyspark.sql import SQLContext
from pyspark.sql.types import *
sqlContext = SQLContext(sc) # SparkContext will be sc by default 

# Read the dataset of your choice (Already loaded with schema)
Data = sqlContext.read.csv("/path", header = True/False, schema = "infer", sep = "delimiter")

# For instance the data has 30 columns from col1, col2, ... col30. If you want to add a 31st column, you can do so by the following:
Data = Data.withColumn("col31", "Code goes here")

# Check the change 
Data.printSchema()

Możemy dodać dodatkowe kolumny bezpośrednio do DataFrame, wykonując poniższe czynności:

from pyspark.sql.functions import when
df = spark.createDataFrame([["amit", 30], ["rohit", 45], ["sameer", 50]], ["name", "age"])
df = df.withColumn("profile", when(df.age >= 40, "Senior").otherwise("Executive"))
df.show()