Spark 2 Dataframe с использованием scala

Question

Фрейм данных 1

Person  Work_order  email  
P1  111 123@gmail.com  
P2  222 123@gmail.com   
P3  111 123@gmail.com   
P4  -1  123@gmail.com   
P5  444 999@gmail.com   

val person = Seq(
  ("P1", "111", "123@gmail.com"),
  ("P2", "222", "123@gmail.com"),
  ("P3", "111", "123@gmail.com"),
  ("P4", "-1", "123@gmail.com"),
  ("P5", "444", "999@gmail.com")).toDF("person", "work_order_person", "email_person")

Фрейм данных 2

Work_order  email   
111 123@gmail.com   
222 123@gmail.com    
444 999@gmail.com 

val workOrder = Seq(
  ("111", "123@gmail.com"),
  ("222", "123@gmail.com"),
  ("444", "999@gmail.com")).toDF("work_order", "email")

С двумя вышеупомянутыми наборами данных [Рабочий заказ и Персона]Я должен построить этот отчет ниже.Любая идея, как это будет сделано.

Отчет
Work_order email Total_count_aggregation_on_the_email Count_exclusive_the_self_work_order_id
111 123@gmail.com 4 2
222 123@gmail.com 4 3
444 999@gmail.com 1 0

Ниже один не является частью отчета.это просто объяснение количества отчетов

Total_count_aggregation_on_the_email

4 = [123@gmail.com,123@gmail.com,123@gmail.com,123@gmail.com]
4 = [123@gmail.com,123@gmail.com,123@gmail.com,123@gmail.com]
1 = [999@gmail.com]

Count_exclusive_the_self_work_order_id

 2 = [222   123@gmail.com ,-1   123@gmail.com]
 3 = [111   123@gmail.com, 111  123@gmail.com, -1   123@gmail.com ]
 0 = []

Возможно ли это с фреймами данных для создания таких отчетов.Ценю твою помощь.