我有一個資料框（df1），看起來像這樣：
-------------------------
|foods |
-------------------------
|[Apple, Apple, Banana] |
|[Apple, Carrot, Broccoli]|
|[Spinach] |
-------------------------
我想在另一個看起來像這樣的資料框（df2）中進行查找：
------------------
|food |category |
------------------
|Apple |Fruit |
|Carrot |Vegetable|
|Broccoli|Vegetable|
|Banana |Fruit |
|Spinach |Vegetable|
------------------
并使生成的資料框如下所示：
------------------------- ----------------------------- ---------
|foods |categories |has fruit|
------------------------- ----------------------------- ---------
|[Apple, Apple, Banana] |[Fruit, Fruit, Fruit] |true |
|[Apple, Carrot, Broccoli]|[Fruit, Vegetable, Vegetable]|true |
|[Spinach] |[Vegetable] |false |
------------------------- ----------------------------- ---------
我怎么能在 Spark/Scala 中做到這一點？我是 Scala 的新手，所以對代碼的解釋也可能會有所幫助。謝謝！
這是我目前正在使用的代碼，但出現org.apache.spark.SparkException: Task not serializable錯誤Caused by: java.io.NotSerializableException: org.apache.spark.sql.Column。
var schema = df1.schema("foods").dataType
def func = udf((x: Seq[String]) => {
x.map(x => df2.filter(col("food") === x).select(col("category")).head().getString(0))
}, schema)
df1.withColumn("categories", func($"foods")).show()
我會很感激一些幫助。代碼不需要干凈。謝謝你。
我試過把 df2 變成一個地圖，并稍微改變了代碼：
var mappy = df2.map{ r => (r.getString(0), r.getString(1))}.collect.toMap
var schema = df1.schema("foods").dataType
def func = udf((x: Seq[String]) => {
x.map(x => mappy.getOrElse(x, ""))
}, schema)
df1.withColumn("categories", func($"foods")).show()
但是，現在我收到此錯誤：
java.lang.StackOverflowError
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1189)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
(((repeats)))
抱歉代碼亂七八糟。這是為了分析，而不是生產。再次感謝！
uj5u.com熱心網友回復：
我像這樣準備了您的輸入資料幀：
// below import is needed for you to use toDF() method to create DataFrames on the fly.
import spark.implicits._
val df1 = List(
(List("Apple", "Apple", "Banana")),
(List("Apple", "Carrot", "Broccoli")),
(List("Spinach"))
).toDF("foods")
val df2 = List(
("Apple", "Fruit"),
("Carrot", "Vegetable"),
("Broccoli", "Vegetable"),
("Banana", "Fruit"),
("Spinach", "Vegetable")
).toDF("food", "category")
我使用 DataFrames 和使用 groupBy 的聚合函式的簡單解決方案可以為您提供所需的輸出（如下所示）：
// imports
import org.apache.spark.sql.functions._
// code
val df1_altered = df1
// explode() : creates a new row for every element in the Array
.withColumn("each_food_from_list", explode(col("foods")))
// df1_altered.show()
val df2_altered = df2
.withColumn(
"has_fruit",
when(col("category").equalTo(lit("Fruit")), true).otherwise(false)
)
// df2_altered.show()
df1_altered
.join(df2_altered, df1_altered("each_food_from_list") === df2_altered("food"), "inner")
//groupBy() : acts as the opposite of explode() by grouping multiple rows together as one based on a column; with the specified mandatory Aggregate function(s)
.groupBy(col("foods"))
.agg(
collect_list(col("category")) as "categories",
max(col("has_fruit")) as "has fruit"
)
.show(false)
// ------------------------- ----------------------------- ---------
// |foods |categories |has fruit|
// ------------------------- ----------------------------- ---------
// |[Apple, Apple, Banana] |[Fruit, Fruit, Fruit] |true |
// |[Apple, Carrot, Broccoli]|[Fruit, Vegetable, Vegetable]|true |
// |[Spinach] |[Vegetable] |false |
// ------------------------- ----------------------------- ---------
編輯：在第一個 DataFrame 中存在重復項時，您可以使用生成的 ID 列，然后使用兩個列進行 groupBy
已經提到了下面的3行代碼更改為change1，change2和change3：
// change1
val df1_with_id = df1.withColumn("id", monotonically_increasing_id())
// change2
val df1_altered = df1_with_id
.withColumn("each_food_from_list", explode(col("foods")))
// df1_altered.show()
val df2_altered = df2
.withColumn(
"has_fruit",
when(col("category").equalTo(lit("Fruit")), true).otherwise(false)
)
// df2_altered.show()
df1_altered
.join(df2_altered, df1_altered("each_food_from_list") === df2_altered("food"), "inner")
// change3
.groupBy(col("id"),col("foods"))
.agg(
collect_list(col("category")) as "categories",
max(col("has_fruit")) as "has fruit"
)
.show(false)
// --- ------------------------- ----------------------------- ---------
//|id |foods |categories |has fruit|
// --- ------------------------- ----------------------------- ---------
//|0 |[Apple, Apple, Banana] |[Fruit, Fruit, Fruit] |true |
//|1 |[Apple, Apple, Banana] |[Fruit, Fruit, Fruit] |true |
//|3 |[Spinach] |[Vegetable] |false |
//|2 |[Apple, Carrot, Broccoli]|[Fruit, Vegetable, Vegetable]|true |
// --- ------------------------- ----------------------------- ---------
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/432891.html