时间戳

首页 > 解决方案 > 使用来自另一个数据集的值搜索和更新 Spark 数据集列

java - 使用来自另一个数据集的值搜索和更新 Spark 数据集列

问题描述

Java 8 和 Spark 2.11:2.3.2 在这里。虽然我更喜欢 Java API 答案，但我确实会讲一点 Scala，所以我将能够理解其中提供的任何答案！但是如果可能的话，Java（请）！

我有两个具有不同模式的数据集，除了一个常见的“ model_number”（字符串）列：两者都存在。

对于我的第一个数据集中的每一行（我们称之为d1），我需要扫描/搜索第二个数据集（“ d2”）以查看是否存在具有相同的行model_number，如果是，则更新另一d2列。

这是我的数据集架构：

d1
===========
model_number : string
desc : string
fizz : string
buzz : date

d2
===========
model_number : string
price : double
source : string

再说一次，如果d1一行有 a model_numberof ，比如 12345，并且d2一行也有相同的model_number，我想d2.price通过将它乘以来更新10.0。

迄今为止我最好的尝试：

// I *think* this would give me a 3rd dataset with all d1 and d2 columns, but only
// containing rows from d1 and d2 that have matching 'model_number' values
Dataset<Row> d3 = d1.join(d2, d1.col("model_number") == d2.col("model_number"));

// now I just need to update d2.price based on matching
Dataset<Row> d4 = d3.withColumn("adjusted_price", d3.col("price") * 10.0);

谁能帮我在这里越过终点线？提前致谢！

标签： javaapache-sparkapache-spark-datasetapache-spark-2.0

解决方案

这里有几点，正如评论中提到的@VamsiPrabhala，您需要使用的功能join在您的特定字段上。关于“ update”，您需要记住df,ds和rddinspark是不可变的，所以您不能使用update它们。所以，这里的解决方案是，在join你df的之后，你需要执行你的计算，在这种情况下是乘法，在 aselect或使用withColumnand thenselect中。换句话说，您不能更新该列，但您可以df使用“ new”列创建新的。

例子：

Input data:

+------------+------+------+----+
|model_number|  desc|  fizz|buzz|
+------------+------+------+----+
|     model_a|desc_a|fizz_a|null|
|     model_b|desc_b|fizz_b|null|
+------------+------+------+----+

+------------+-----+--------+
|model_number|price|  source|
+------------+-----+--------+
|     model_a| 10.0|source_a|
|     model_b| 20.0|source_b|
+------------+-----+--------+

使用join将输出：

val joinedDF = d1.join(d2, "model_number")
joinedDF.show()

+------------+------+------+----+-----+--------+
|model_number|  desc|  fizz|buzz|price|  source|
+------------+------+------+----+-----+--------+
|     model_a|desc_a|fizz_a|null| 10.0|source_a|
|     model_b|desc_b|fizz_b|null| 20.0|source_b|
+------------+------+------+----+-----+--------+

应用您的计算：

joinedDF.withColumn("price", col("price") * 10).show()

output:
+------------+------+------+----+-----+--------+
|model_number|  desc|  fizz|buzz|price|  source|
+------------+------+------+----+-----+--------+
|     model_a|desc_a|fizz_a|null| 100.0|source_a|
|     model_b|desc_b|fizz_b|null| 200.0|source_b|
+------------+------+------+----+-----+--------+

推荐阅读