Merge two columns in a single DataFrame and count the occurrences using PySpark

You can use explode with array fuction to merge the columns into one then simply group by and count, like this :

from pyspark.sql.functions import col, array, explode, count

df.select(explode(array("name1", "name2")).alias("name")) \
    .filter("nullif(name, '') is not null") \
    .groupBy("name") \
    .agg(count("*").alias("Occurrence")) \
    .show()

#+-----------------+----------+
#|             name|Occurrence|
#+-----------------+----------+
#|Meindert I Tholen|         2|
#|    Jeanny  Thorn|         3|
#|       Luc  Krier|         2|
#|  Teddy E Beecher|         1|
#|Philippe  Schauss|         1|
#|       Ben Weller|         1|
#|      Liam Muller|         1|
#+-----------------+----------+

Another way is to select each column, union then group by and count:

df.select(col("name1").alias("name")).union(df.select(col("name2").alias("name"))) \
    .filter("nullif(name, '') is not null")\
    .groupBy("name") \
    .agg(count("name").alias("Occurrence")) \
    .show()

Many fancy answers out there, but the easiest solution should be to do a union and then aggregate the count:

df2 = (df.select('name1')
         .union(df.select('name2'))
         .filter("name1 != ''")
         .groupBy('name1')
         .count()
         .toDF('name', 'Occurrence')
      )

df2.show()
+-----------------+----------+
|             name|Occurrence|
+-----------------+----------+
|Meindert I Tholen|         2|
|    Jeanny  Thorn|         3|
|       Luc  Krier|         2|
|  Teddy E Beecher|         1|
|Philippe  Schauss|         1|
|       Ben Weller|         1|
|      Liam Muller|         1|
+-----------------+----------+

There are better ways to do it. One naive way of doing it is as follows

from collections import Counter

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("OccurenceCount").getOrCreate()

df = spark.createDataFrame([
 ["Luc  Krier","Jeanny  Thorn"],
 ["Jeanny  Thorn","Ben Weller"],
 [ "Teddy E Beecher","Luc  Krier"],
 ["Philippe  Schauss","Jeanny  Thorn"],
 ["Meindert I Tholen","Liam Muller"],
 ["Meindert I Tholen",""]
]).toDF("name1", "name2")


counter_dict = dict(Counter(df.select("name1", "name2").rdd.flatMap(lambda x: x).collect()))

counter_list = list(map(list, counter_dict.items()))

frequency_df = spark.createDataFrame(counter_list, ["name", "Occurrence"])

frequency_df.show()

Output:

+-----------------+----------+
|             name|Occurrence|
+-----------------+----------+
|                 |         1|
|      Liam Muller|         1|
|  Teddy E Beecher|         1|
|       Ben Weller|         1|
|    Jeanny  Thorn|         3|
|       Luc  Krier|         2|
|Philippe  Schauss|         1|
|Meindert I Tholen|         2|
+-----------------+----------+

Does this work?

# Groupby & count both dataframes individually to reduce size.
df_name1 = (df.groupby(['name1']).count()
            .withColumnRenamed('name1', 'name')
            .withColumnRenamed('count', 'count1'))
df_name2 = (df.groupby(['name2']).count()
            .withColumnRenamed('name2', 'name')
            .withColumnRenamed('count', 'count2'))

# Join the two dataframes containing frequency counts
# Any null value in the 'count' column can be correctly interpreted as zero.
df_count = (df_name1.join(df_name2, on=['name'], how='outer')
            .fillna(0, subset=['count1', 'count2']))

# Sum the two counts and drop the useless columns
df_count = (df_count.withColumn('count', df_count['count1'] + df_count['count2'])
            .drop('count1').drop('count2').dropna(subset=['name']))

# (Optional) While any rows with a null name have been removed, rows with an
# empty string ("") for a name are still there. We can drop the empty name
# rows like this.
df_count = df_count[df_count['name'] != '']
df_count.show()
# +-----------------+-----+
# |             name|count|
# +-----------------+-----+
# |Meindert I Tholen|    2|
# |    Jeanny  Thorn|    3|
# |       Luc  Krier|    2|
# |  Teddy E Beecher|    1|
# |Philippe  Schauss|    1|
# |       Ben Weller|    1|
# |      Liam Muller|    1|
# +-----------------+-----+

You can get the required output as follows in scala :

import org.apache.spark.sql.functions._
val df = Seq(
 ("Luc  Krier","Jeanny  Thorn"),
 ("Jeanny  Thorn","Ben Weller"),
 ( "Teddy E Beecher","Luc  Krier"),
 ("Philippe  Schauss","Jeanny  Thorn"),
 ("Meindert I Tholen","Liam Muller"),
 ("Meindert I Tholen","")
).toDF("name1", "name2")

val df1 = df.filter($"name1".isNotNull).filter($"name1" !== 
"").groupBy("name1").agg(count("name1").as("count1"))
val df2 = df.filter($"name2".isNotNull).filter($"name2" !== 
"").groupBy("name2").agg(count("name2").as("count2"))
val newdf = df1.join(df2, $"name1" === $"name2","outer").withColumn("count1", 
when($"count1".isNull,0).otherwise($"count1")).withColumn("count2", 
when($"count2".isNull,0).otherwise($"count2")).withColumn("Count",$"count1" + 
$"count2")
val finalDF =newdf.withColumn("name",when($"name1".isNull,$"name2")
.when($"name2".isNull,$"name1").otherwise($"name1")).select("name","Count")
display(finalDF)

You can see the final output as image below :