Làm thế nào để thêm một cột không đổi trong một DataFrame Spark?

Question

Tôi muốn thêm cột vào một số DataFrame với một số giá trị tùy ý (giống nhau cho mỗi hàng). Tôi nhận được một lỗi khi tôi sử dụng withColumn như sau:

dt.withColumn('new_column', 10).head(5) 

--------------------------------------------------------------------------- 
AttributeError       Traceback (most recent call last) 
<ipython-input-50-a6d0257ca2be> in <module>() 
     1 dt = (messages 
     2  .select(messages.fromuserid, messages.messagetype, floor(messages.datetime/(1000*60*5)).alias("dt"))) 
----> 3 dt.withColumn('new_column', 10).head(5) 

/Users/evanzamir/spark-1.4.1/python/pyspark/sql/dataframe.pyc in withColumn(self, colName, col) 
    1166   [Row(age=2, name=u'Alice', age2=4), Row(age=5, name=u'Bob', age2=7)] 
    1167   """ 
-> 1168   return self.select('*', col.alias(colName)) 
    1169 
    1170  @ignore_unicode_prefix 

AttributeError: 'int' object has no attribute 'alias' 

Dường như tôi có thể đánh lừa các chức năng vào làm việc như tôi muốn bằng cách thêm và trừ đi một trong những cột khác (do đó họ thêm vào zero) và sau đó thêm số tôi muốn (10 trong trường hợp này):

dt.withColumn('new_column', dt.messagetype - dt.messagetype + 10).head(5) 

[Row(fromuserid=425, messagetype=1, dt=4809600.0, new_column=10), 
Row(fromuserid=47019141, messagetype=1, dt=4809600.0, new_column=10), 
Row(fromuserid=49746356, messagetype=1, dt=4809600.0, new_column=10), 
Row(fromuserid=93506471, messagetype=1, dt=4809600.0, new_column=10), 
Row(fromuserid=80488242, messagetype=1, dt=4809600.0, new_column=10)] 

Điều này cực kỳ đáng sợ, phải không? Tôi cho rằng có một cách hợp pháp hơn để làm điều này?

Answer 1

Spark 2.2+

Spark 2.2 giới thiệu typedLit để hỗ trợ Seq, Map, và Tuples (SPARK-19254) và sau đây gọi cần được hỗ trợ (Scala):

import org.apache.spark.sql.functions.typedLit 

df.withColumn("some_array", typedLit(Seq(1, 2, 3))) 
df.withColumn("some_struct", typedLit(("foo", 1, .0.3))) 
df.withColumn("some_map", typedLit(Map("key1" -> 1, "key2" -> 2))) 

Spark 1.3+ (lit), 1.4+ (array, struct), 2.0+ (map):

Đối số thứ hai cho DataFrame.withColumn phải là một Column vì vậy bạn phải sử dụng một chữ:

from pyspark.sql.functions import lit 

df.withColumn('new_column', lit(10)) 

Nếu bạn cần cột phức tạp bạn có thể xây dựng các khối sử dụng như array:

from pyspark.sql.functions import array, create_map, struct 

df.withColumn("some_array", array(lit(1), lit(2), lit(3))) 
df.withColumn("some_struct", struct(lit("foo"), lit(1), lit(.3))) 
df.withColumn("some_map", create_map(lit("key1"), lit(1), lit("key2"), lit(2))) 

Chính xác cùng một phương pháp có thể được sử dụng trong Scala.

import org.apache.spark.sql.functions.{array, lit, map, struct} 

df.withColumn("new_column", lit(10)) 
df.withColumn("map", map(lit("key1"), lit(1), lit("key2"), lit(2))) 

Cũng có thể, mặc dù chậm hơn, để sử dụng UDF.

Answer 2

Trong spark 2.2 có hai cách để thêm giá trị không đổi trong một cột trong DataFrame:

1) Sử dụng lit

2) Sử dụng typedLit.

Sự khác biệt giữa hai là typedLit cũng có thể xử lý các loại scala được tham số, ví dụ:Danh sách, Seq, và Bản đồ

mẫu DataFrame:

val df = spark.createDataFrame(Seq((0,"a"),(1,"b"),(2,"c"))).toDF("id", "col1") 

+---+----+ 
| id|col1| 
+---+----+ 
| 0| a| 
| 1| b| 
+---+----+ 

1) Sử dụng lit: Thêm chuỗi giá trị liên tục trong cột mới có tên newcol:

import org.apache.spark.sql.functions.lit 
val newdf = df.withColumn("newcol",lit("myval")) 

Kết quả:

+---+----+------+ 
| id|col1|newcol| 
+---+----+------+ 
| 0| a| myval| 
| 1| b| myval| 
+---+----+------+ 

2) Sử dụng typedLit:

import org.apache.spark.sql.functions.typedLit 
df.withColumn("newcol", typedLit(("sample", 10, .044))) 

Kết quả:

+---+----+-----------------+ 
| id|col1|   newcol| 
+---+----+-----------------+ 
| 0| a|[sample,10,0.044]| 
| 1| b|[sample,10,0.044]| 
| 2| c|[sample,10,0.044]| 
+---+----+-----------------+