Làm cách nào để cập nhật cột dựa trên điều kiện (giá trị trong một nhóm)?

Question

tôi có một df

+---+----+-----+ 
|sno|dept|color| 
+---+----+-----+ 
| 1| fn| red| 
| 2| fn| blue| 
| 3| fn|green| 
+---+----+-----+ 

Nếu bất kỳ giá trị cột màu đỏ sau đó tôi nên cập nhật tất cả các giá trị cột như đỏ

như dưới đây

+---+----+-----+ 
|sno|dept|color| 
+---+----+-----+ 
| 1| fn| red| 
| 2| fn| red| 
| 3| fn| red| 
+---+----+-----+ 

tôi không thể tìm nó ra . Bất kỳ sự giúp đỡ xin vui lòng tôi đã mệt mỏi sau đang

val gp=jdbcDF.filter($"dept".contains("fn")) 
    //.withColumn("newone",when($"dept"==="fn","RED").otherwise("NULL")) 
    gp.show() 
gp.map(
    row=>{ 
    val row1=row.getAs[String](1) 
    var row2=row.getAs[String](2) 
    val make=if(row1 =="fn") row2="red" 
    Row(row(0),row(1),make) 
    } 
).collect().foreach(println) 

Answer 1

Given:

val df = Seq(
    (1, "fn", "red"), 
    (2, "fn", "blue"), 
    (3, "fn", "green"), 
    (4, "aa", "blue"), 
    (5, "aa", "green"), 
    (6, "bb", "red"), 
    (7, "bb", "red"), 
    (8, "aa", "blue") 
).toDF("id", "fn", "color") 

Do việc tính toán:

val redOrNot = df.groupBy("fn") 
    .agg(collect_set('color) as "values") 
    .withColumn("hasRed", array_contains('values, "red")) 

// gives null for no option 
val colorPicker = when('hasRed, "red") 
val result = df.join(redOrNot, "fn") 
    .withColumn("resultColor", colorPicker) 
    .withColumn("color", coalesce('resultColor, 'color)) // skips nulls that leads to the answer 
    .select('id, 'fn, 'color) 

Các result trông như sau (mà dường như là một câu trả lời):

scala> result.show 
+---+---+-----+ 
| id| fn|color| 
+---+---+-----+ 
| 1| fn| red| 
| 2| fn| red| 
| 3| fn| red| 
| 4| aa| blue| 
| 5| aa|green| 
| 6| bb| red| 
| 7| bb| red| 
| 8| aa| blue| 
+---+---+-----+ 

Bạn có thể chuỗi các toán tử when và có giá trị mặc định bằng otherwise. Tham khảo scaladoc of when operator. Tôi nghĩ bạn có thể làm điều gì đó rất giống nhau (và có lẽ hiệu quả hơn) bằng cách sử dụng toán tử cửa sổ hoặc hàm tổng hợp do người dùng xác định (UDAF), nhưng ... well ... hiện không biết cách thực hiện. Để lại nhận xét ở đây để truyền cảm hứng cho người khác ;-)

p.s. Đã học rất nhiều! Cảm ơn ý tưởng!

Answer 2

giải pháp hiệu quả mà không yêu cầu nhóm đắt:

// All groups with `red` 
df.where($"color" === "red").select($"fn".alias("fn_")).distinct 
    // Join with input 
    .join(df.as("df"), $"fn" === $"fn_", "rightouter") 
    // Replace `color` 
    .withColumn("color", when($"fn_"isNull, $"color").otherwise(lit("red"))) 
    .drop("fn_") 

Answer 3

Bạn đang có điều kiện cập nhật DataFrame nếu nó đáp ứng một tài sản nhất định. Trong trường hợp này, thuộc tính là "cột màu chứa" màu đỏ "". Cách thành ngữ để diễn đạt điều này là lọc với vị từ mong muốn và sau đó xác định xem có bất kỳ hàng nào thỏa mãn nó hay không. Không cần tham gia.

import org.apache.spark.sql.functions.lit 
import org.apache.spark.sql.DataFrame 

def makeAllRedIfAnyAreRed(df: DataFrame) = { 
    val containsRed = df.filter(df("color") === "red").count() > 0 
    if (containsRed) df.withColumn("color", lit("red")) else df 
} 

Answer 4

Như có thể có vài hàng trong dataframe lọc tôi thêm giải pháp với isin() và .withColumn() kết hợp.

mẫu DataFrame

val df = Seq(
    (1, "fn", "red"), 
    (2, "fn", "blue"), 
    (3, "fn", "green"), 
    (4, "aa", "blue"), 
    (5, "aa", "green"), 
    (6, "bb", "red"), 
    (7, "bb", "red"), 
    (8, "aa", "blue") 
).toDF("id", "dept", "color") 

Bây giờ Hãy chọn chỉ dept s mà có ít nhất một đỏcolor hàng và đặt nó vào broadcast biến như dưới đây.

val depts = sc.broadcast(df.filter($"color" === "red").select(collect_set("dept")).first.getSeq[String](0))) 

Cập nhật đỏ màu cho lọc depts hồ sơ.

isin() mất một vararg để chuyển đổi danh sách để vararg (depts.value:_*)

//creating new column by giving diff name (clr) to see the diff 
val result = df.withColumn("clr", when($"dept".isin(depts.value:_*),lit("red")) 
        .otherwise($"color")) 

result.show() 

+---+----+-----+-----+ 
| id|dept|color| clr| 
+---+----+-----+-----+ 
| 1| fn| red| red| 
| 2| fn| blue| red| 
| 3| fn|green| red| 
| 4| aa| blue| blue| 
| 5| aa|green|green| 
| 6| bb| red| red| 
| 7| bb| red| red| 
| 8| aa| blue| blue| 
+---+----+-----+-----+ 

Answer 5

Spark 2.2.0: mẫu Dataframe (lấy từ ví dụ trên)

val df = Seq(
    (1, "fn", "red"), 
    (2, "fn", "blue"), 
    (3, "fn", "green"), 
    (4, "aa", "blue"), 
    (5, "aa", "green"), 
    (6, "bb", "red"), 
    (7, "bb", "red"), 
    (8, "aa", "blue") 
).toDF("id", "dept", "color") 

tạo ra một UDF để thực hiện cập nhật bằng cách kiểm tra điều kiện.

val replace_val = udf((x: String,y:String) => if (Option(x).getOrElse("").equalsIgnoreCase("fn")&&(!y.equalsIgnoreCase("red"))) "red" else y) 

val final_df = df.withColumn("color", replace_val($"dept",$"color")) 
final_df.show() 

đầu ra:

spark 1.6:

val conf = new SparkConf().setMaster("local").setAppName("My app") 
val sc = new SparkContext(conf) 
val sqlContext = new SQLContext(sc) 

import sqlContext.implicits._ 
// For implicit conversions like converting RDDs to DataFrames 
val df = sc.parallelize(Seq(
    (1, "fn", "red"), 
    (2, "fn", "blue"), 
    (3, "fn", "green"), 
    (4, "aa", "blue"), 
    (5, "aa", "green"), 
    (6, "bb", "red"), 
    (7, "bb", "red"), 
    (8, "aa", "blue") 
)).toDF("id","dept","color") 


val replace_val = udf((x: String,y:String) => if (Option(x).getOrElse("").equalsIgnoreCase("fn")&&(!y.equalsIgnoreCase("red"))) "red" else y) 
val final_df = df.withColumn("color", replace_val($"dept",$"color")) 

final_df.show()