Adding a new column in Data Frame derived from other columns (Spark)
You have the following possibilities to add a new column:
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame([[1, 2], [3, 4]], ['col1', 'col2'])
df.show()
+----+----+
|col1|col2|
+----+----+
| 1| 2|
| 3| 4|
+----+----+
-- Using the method withColumn
:
import pyspark.sql.functions as F
df.withColumn('col3', F.col('col2') - F.col('col1')) # col function
df.withColumn('col3', df['col2'] - df['col1']) # bracket notation
df.withColumn('col3', df.col2 - df.col1) # dot notation
-- Using the method select
:
df.select('*', (F.col('col2') - F.col('col1')).alias('col3'))
The expression '*'
returns all columns.
-- Using the method selectExpr
:
df.selectExpr('*', 'col2 - col1 as col3')
-- Using SQL:
df.createOrReplaceTempView('df_view')
spark.sql('select *, col2 - col1 as col3 from df_view')
Result:
+----+----+----+
|col1|col2|col3|
+----+----+----+
| 1| 2| 1|
| 3| 4| 1|
+----+----+----+
Additionally, we can use udf
from pyspark.sql.functions import udf,col
from pyspark.sql.types import IntegerType
from pyspark import SparkContext
from pyspark.sql import SQLContext
sc = SparkContext()
sqlContext = SQLContext(sc)
old_df = sqlContext.createDataFrame(sc.parallelize(
[(0, 1), (1, 3), (2, 5)]), ('col_1', 'col_2'))
function = udf(lambda col1, col2 : col1-col2, IntegerType())
new_df = old_df.withColumn('col_n',function(col('col_1'), col('col_2')))
new_df.show()
One way to achieve that is to use withColumn
method:
old_df = sqlContext.createDataFrame(sc.parallelize(
[(0, 1), (1, 3), (2, 5)]), ('col_1', 'col_2'))
new_df = old_df.withColumn('col_n', old_df.col_1 - old_df.col_2)
Alternatively you can use SQL on a registered table:
old_df.registerTempTable('old_df')
new_df = sqlContext.sql('SELECT *, col_1 - col_2 AS col_n FROM old_df')