How to zip two (or more) DataFrame in Spark

by

Operation like this is not supported by a DataFrame API. It is possible to zip two RDDs but to make it work you have to match both number of partitions and number of elements per partition. Assuming this is the case:

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField, StructType, LongType}

val a: DataFrame = sc.parallelize(Seq(
  ("abc", 123), ("cde", 23))).toDF("column_1", "column_2")
val b: DataFrame = sc.parallelize(Seq(Tuple1(1), Tuple1(2))).toDF("column_3")

// Merge rows
val rows = a.rdd.zip(b.rdd).map{
  case (rowLeft, rowRight) => Row.fromSeq(rowLeft.toSeq ++ rowRight.toSeq)}

// Merge schemas
val schema = StructType(a.schema.fields ++ b.schema.fields)

// Create new data frame
val ab: DataFrame = sqlContext.createDataFrame(rows, schema)

If above conditions are not met the only option that comes to mind is adding an index and join:

def addIndex(df: DataFrame) = sqlContext.createDataFrame(
  // Add index
  df.rdd.zipWithIndex.map{case (r, i) => Row.fromSeq(r.toSeq :+ i)},
  // Create schema
  StructType(df.schema.fields :+ StructField("_index", LongType, false))
)

// Add indices
val aWithIndex = addIndex(a)
val bWithIndex = addIndex(b)

// Join and clean
val ab = aWithIndex
  .join(bWithIndex, Seq("_index"))
  .drop("_index")

More Related Contents:

  1. How to query JSON data column using Spark DataFrames?
  2. How to define partitioning of DataFrame?
  3. How do I detect if a Spark DataFrame has a column
  4. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  5. How to split a dataframe into dataframes with same column values?
  6. Provide schema while reading csv file as a dataframe
  7. Dropping a nested column from Spark DataFrame
  8. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  9. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  10. Joining Spark dataframes on the key
  11. Apache Spark how to append new column from list/array to Spark dataframe
  12. Create new Dataframe with empty/null field values
  13. Provide schema while reading csv file as a dataframe in Scala Spark
  14. Renaming column names of a DataFrame in Spark Scala
  15. Derive multiple columns from a single column in a Spark DataFrame
  16. Replace missing values with mean – Spark Dataframe
  17. What is going wrong with `unionAll` of Spark `DataFrame`?
  18. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  19. How to get keys and values from MapType column in SparkSQL DataFrame
  20. Spark DataFrames when udf functions do not accept large enough input variables
  21. Filter spark DataFrame on string contains
  22. Spark – load CSV file as DataFrame?
  23. SparkSQL: How to deal with null values in user defined function?
  24. How to aggregate values into collection after groupBy?
  25. MatchError while accessing vector column in Spark 2.0
  26. Perform a typed join in Scala with Spark Datasets
  27. DataFrame-ified zipWithIndex
  28. Why is join not possible after show operator?
  29. How to create DataFrame from Scala’s List of Iterables?
  30. How to create a Dataset of Maps?

Leave a Comment