Spark Scala: Cách chuyển đổi một cột trong DF

Question

Tôi có một khung dữ liệu trong Spark với nhiều cột và một udf mà tôi đã xác định. Tôi muốn cùng một dataframe trở lại, ngoại trừ với một cột chuyển đổi. Hơn nữa, udf của tôi lấy một chuỗi và trả về dấu thời gian. Có cách nào làm dễ hơn không? Tôi đã thử

val test = myDF.select("my_column").rdd.map(r => getTimestamp(r)) 

nhưng điều này trả về RDD và chỉ với cột được chuyển đổi.

Answer 1

Nếu bạn thực sự cần phải sử dụng chức năng của bạn, tôi có thể đề nghị hai lựa chọn:

1) Sử dụng bản đồ/toDF:

import org.apache.spark.sql.Row 
import sqlContext.implicits._ 

def getTimestamp: (String => java.sql.Timestamp) = // your function here 

val test = myDF.select("my_column").rdd.map { 
    case Row(string_val: String) => (string_val, getTimestamp(string_val)) 
}.toDF("my_column", "new_column") 

2) Sử dụng UDFs (UserDefinedFunction):

import org.apache.spark.sql.functions._ 

def getTimestamp: (String => java.sql.Timestamp) = // your function here 

val newCol = udf(getTimestamp).apply(col("my_column")) // creates the new column 
val test = myDF.withColumn("new_column", newCol) // adds the new column to original DF 

Có thêm chi tiết về Spark SQL UDFs trong this nice article by Bill Chambers.

---

Ngoài ra,

Nếu bạn chỉ muốn chuyển một cột StringType vào một cột TimestampType bạn có thể sử dụng unix_timestampcolumn function sẵn từ Spark SQL 1.5:

val test = myDF 
    .withColumn("new_column", unix_timestamp(col("my_column"), "yyyy-MM-dd HH:mm").cast("timestamp")) 

Lưu ý: Đối với spark 1.5.x, cần nhân kết quả của unix_timestamp trước 1000 trước khi truyền tới dấu thời gian (iss ue SPARK-11724). Mã kết quả sẽ là:

val test = myDF 
    .withColumn("new_column", (unix_timestamp(col("my_column"), "yyyy-MM-dd HH:mm") *1000L).cast("timestamp")) 

Edit: Thêm tùy chọn udf