Spark Dataset API - присоединиться

Чтобы использовать joinWith, вам сначала нужно создать DataSet, а скорее всего два из них. Чтобы создать DataSet, вам нужно создать класс case, соответствующий вашей схеме, и вызвать DataFrame.as[T], где T - ваш класс case. Так:

case class KeyValue(key: Int, value: String)
val df = Seq((1,"asdf"),(2,"34234")).toDF("key", "value")
val ds = df.as[KeyValue]
// org.apache.spark.sql.Dataset[KeyValue] = [key: int, value: string]

Вы также можете пропустить класс case и использовать кортеж:

val tupDs = df.as[(Int,String)]
// org.apache.spark.sql.Dataset[(Int, String)] = [_1: int, _2: string]

Затем, если у вас был другой класс case / DF, например, скажите:

case class Nums(key: Int, num1: Double, num2: Long)
val df2 = Seq((1,7.7,101L),(2,1.2,10L)).toDF("key","num1","num2")
val ds2 = df2.as[Nums]
// org.apache.spark.sql.Dataset[Nums] = [key: int, num1: double, num2: bigint]

Тогда, хотя синтаксис join и joinWith похож, результаты разные:

df.join(df2, df.col("key") === df2.col("key")).show
// +---+-----+---+----+----+
// |key|value|key|num1|num2|
// +---+-----+---+----+----+
// |  1| asdf|  1| 7.7| 101|
// |  2|34234|  2| 1.2|  10|
// +---+-----+---+----+----+

ds.joinWith(ds2, df.col("key") === df2.col("key")).show
// +---------+-----------+
// |       _1|         _2|
// +---------+-----------+
// | [1,asdf]|[1,7.7,101]|
// |[2,34234]| [2,1.2,10]|
// +---------+-----------+

Как видите, joinWith оставляет объекты нетронутыми как части кортежа, а join объединяет столбцы в единое пространство имен. (Что вызовет проблемы в приведенном выше случае, потому что имя столбца «ключ» повторяется.)

Как ни странно, я должен использовать df.col("key") и df2.col("key"), чтобы создать условия для присоединения ds и ds2 - если вы используете только col("key") с обеих сторон, это не сработает, а ds.col(...) не существует. Однако использование оригинального df.col("key") помогает.

Из https://docs.cloud.databricks.com/docs/latest/databricks_guide/05%20Spark/1%20Intro%20Datasets.html

похоже, ты мог бы просто сделать

dfA.as("A").joinWith(dfB.as("B"), $"A.date" === $"B.date" )

В приведенном выше примере вы можете попробовать следующее:

Определите класс case для вашего вывода

case class JoinOutput(key:Int, value:String, num1:Double, num2:Long) 

Объедините два набора данных с помощью Seq("key"), это поможет вам избежать двух повторяющихся ключевых столбцов в выходных данных, что также поможет применить класс case или получить данные на следующем шаге.

val joined = ds.join(ds2, Seq("key")).as[JoinOutput]
// res27: org.apache.spark.sql.Dataset[JoinOutput] = [key: int, value: string ... 2 more fields]

Вместо этого результат будет плоским:

joined.show

+---+-----+----+----+
|key|value|num1|num2|
+---+-----+----+----+
|  1| asdf| 7.7| 101|
|  2|34234| 1.2|  10|
+---+-----+----+----+