实现此目的的一种方法是使用join
如果两个数据框中都有一些共同的列,那么您可以对该列执行联接并获得您想要的结果。
Example:
import sparkSession.sqlContext.implicits._
val df1 = Seq((1, "Anu"),(2, "Suresh"),(3, "Usha"), (4, "Nisha")).toDF("id","name")
val df2 = Seq((1, 23),(2, 24),(3, 24), (4, 25), (5, 30), (6, 32)).toDF("id","age")
val df = df1.as("df1").join(df2.as("df2"), df1("id") === df2("id")).select("df1.id", "df1.name", "df2.age")
df.show()
Output:
+---+------+---+
| id| name|age|
+---+------+---+
| 1| Anu| 23|
| 2|Suresh| 24|
| 3| Usha| 24|
| 4| Nisha| 25|
+---+------+---+
Update:
如果您在两个数据框中没有任何共同的唯一 ID,请创建一个并使用它。
import sparkSession.sqlContext.implicits._
import org.apache.spark.sql.functions._
var sourceDf = Seq(1000, 2000, 3000, 4000).toDF("InputGas")
var resultDf = Seq((0, 111, 1111), (0, 222, 2222), (1, 333, 3333), (2, 444, 4444)).toDF("Time", "CalcGas", "Speed")
sourceDf = sourceDf.withColumn("rowId1", monotonically_increasing_id())
resultDf = resultDf.withColumn("rowId2", monotonically_increasing_id())
val df = sourceDf.as("df1").join(resultDf.as("df2"), sourceDf("rowId1") === resultDf("rowId2"), "inner").select("df1.InputGas", "df2.Time", "df2.CalcGas", "df2.Speed")
df.show()
Output:
+--------+----+-------+-----+
|InputGas|Time|CalcGas|Speed|
+--------+----+-------+-----+
| 1000| 0| 111| 1111|
| 2000| 0| 222| 2222|
| 3000| 1| 333| 3333|
| 4000| 2| 444| 4444|
+--------+----+-------+-----+