How to group by common element in array?

by

Include graphframes (the latest supported Spark version is 2.1, but it should support 2.2 as well, if you use newer you’ll have to build your own with 2.3 patch) replacing XXX with Spark version and YYY with Scala version:

spark.jars.packages  graphframes:graphframes:0.5.0-sparkXXX-s_YYY

Add explode keys:

import org.apache.spark.sql.functions._

val df = Seq(
   (Seq("k1", "k2"), "v1"), (Seq("k2"), "v2"),
   (Seq("k3", "k2"), "v3"), (Seq("k4"), "v4")
).toDF("key", "value")

val edges = df.select(
  explode($"key") as "src", $"value" as "dst")

Convert to graphframe:

import org.graphframes._

val gf = GraphFrame.fromEdges(edges)

Set checkpoint directory (if not set):

import org.apache.spark.sql.SparkSession

val path: String = ???
val spark: SparkSession = ???
spark.sparkContext.setCheckpointDir(path)

Find connected components:

val components = GraphFrame.fromEdges(edges).connectedComponents.setAlgorithm("graphx").run

Join result with input data:

 val result = components.where($"id".startsWith("v")).toDF("value", "group").join(df, Seq("value"))

Check result:

result.show

// +-----+------------+--------+
// |value|       group|     key|
// +-----+------------+--------+
// |   v3|489626271744|[k3, k2]|
// |   v2|489626271744|    [k2]|
// |   v4|532575944704|    [k4]|
// |   v1|489626271744|[k1, k2]|
// +-----+------------+--------+

More Related Contents:

  1. Using a column value as a parameter to a spark DataFrame function
  2. Split Spark Dataframe string column into multiple columns
  3. DataFrame join optimization – Broadcast Hash Join
  4. How to control partition size in Spark SQL
  5. Avoid performance impact of a single partition mode in Spark window functions
  6. How to import multiple csv files in a single load?
  7. Convert date from String to Date format in Dataframes
  8. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  9. Spark DataFrame: count distinct values of every column
  10. How to split a list to multiple columns in Pyspark?
  11. pyspark dataframe filter or include based on list
  12. Use collect_list and collect_set in Spark SQL
  13. What is the maximum size for a broadcast object in Spark?
  14. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  15. How to find mean of grouped Vector columns in Spark SQL?
  16. Generate a Spark StructType / Schema from a case class
  17. How does createOrReplaceTempView work in Spark?
  18. Fill in null with previously known good value with pyspark
  19. How do I add an persistent column of row ids to Spark DataFrame?
  20. Keep only duplicates from a DataFrame regarding some field
  21. DataFrame partitionBy to a single Parquet file (per partition)
  22. Temp table caching with spark-sql
  23. Using windowing functions in Spark
  24. How can I access python variable in Spark SQL?
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. How to get Kafka offsets for structured query for manual and reliable offset management?
  27. How to calculate Median in spark sqlContext for column of data type double
  28. Spark: disk I/O on stage boundaries explanation
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment