Объединение нескольких кадров данных по горизонтали - PullRequest
Новая
       20

Объединение нескольких кадров данных по горизонтали

1 голос
/ 29 мая 2019

У меня есть следующий фрейм данных 

val count :Dataframe = spark.sql("select 1,$database_name,$table_name count(*) from $table_name ")

Выход:

1, инвентарь, T076p, 4332 

val dist_count :Dataframe = spark.sql("1,select distinct count(*) from $table_name")`

Выход:

4112 или 4332 (может быть таким же) 

val truecount : Dataframe = spark.sql("select 1,count(*) from $table_name where flag =true")`

Выход:

4330 

   val Falsecount : DataFrame = spark.sql("select 1,count(*) from $table_name where flag =false")

Выход:

4332

Вопрос: Как мне присоединиться выше dataframe, чтобы получить результирующий dataframe, который дает мне вывод.
как показано ниже.

сток, T076p, 4332,4332,4330

Здесь запятая для разделителя столбцов

P.S. Я добавил 1 к каждому dataframe, чтобы я мог использовать объединение dataframes (поэтому 1 здесь не является обязательным.)

1 Ответ

1 голос
/ 30 мая 2019

Вопрос :
Как мне присоединиться к вышеуказанному фрейму данных, чтобы получить результирующий фрейм данных, который дает мне o / p, как показано ниже.

stock, T076p, 4332,4332,4330 - Здесь запятая для разделителя столбцов

, просто посмотрите на этот пример.Я сымитировал ваше требование с фиктивными кадрами данных, как показано ниже. 

package com.examples

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object MultiDFJoin {
  def main(args: Array[String]) {
    import org.apache.spark.sql.functions._
    Logger.getLogger("org").setLevel(Level.OFF)

    val spark = SparkSession.builder.
      master("local")
      .appName(this.getClass.getName)
      .getOrCreate()
    import spark.implicits._
    val columns = Array("column1", "column2", "column3", "column4")
    val df1 = (Seq(
      (1, "stock", "T076p", 4332))
      ).toDF(columns: _*).as("first")
    df1.show()
    val df2 = Seq((1, 4332)).toDF(columns.slice(0, 2): _*).as("second")
    df2.show()
    val df3 = Seq((1, 4330)).toDF(columns.slice(0, 2): _*).as("third")
    df3.show()
    val df4 = Seq((1, 4332)).toDF(columns.slice(0, 2): _*).as("four")
    df4.show()
    val finalcsv = df1.join(df2, col("first.column1") === col("second.column1")).selectExpr("first.*", "second.column2")
      .join(df3, Seq("column1")).selectExpr("first.*", "third.column2")
      .join(df4, Seq("column1"))
      .selectExpr("first.*", "third.column2", "four.column2")
      .drop("column1").collect.mkString(",") // this column used for just joining hence dropping
    print(finalcsv)
  }
}

Результат: 

+-------+-------+-------+-------+
|column1|column2|column3|column4|
+-------+-------+-------+-------+
|      1|  stock|  T076p|   4332|
+-------+-------+-------+-------+

+-------+-------+
|column1|column2|
+-------+-------+
|      1|   4332|
+-------+-------+

+-------+-------+
|column1|column2|
+-------+-------+
|      1|   4330|
+-------+-------+

+-------+-------+
|column1|column2|
+-------+-------+
|      1|   4332|
+-------+-------+

[stock,T076p,4332,4330,4332]
...