бросить повторяющиеся слова в длинной строке, используя Scala

Question

Мне любопытно узнать, как отбрасывать повторяющиеся слова в строках, которые содержатся в столбце данных.Я хотел бы сделать это с помощью Scala.В качестве примера ниже вы можете найти фрейм данных, который я хотел бы преобразовать.

dataframe:

val dataset1 = Seq(("66", "a,b,c,a", "4"), ("67", "a,f,g,t", "0"), ("70", "b,b,b,d", "4")).toDF("KEY1", "KEY2", "ID") 

+----+-------+---+
|KEY1|   KEY2| ID|
+----+-------+---+
|  66|a,b,c,a|  4|
|  67|a,f,g,t|  0|
|  70|b,b,b,d|  4|
+----+-------+---+

результат:

+----+----------+---+
|KEY1|      KEY2| ID|
+----+----------+---+
|  66|   a, b, c|  4|
|  67|a, f, g, t|  0|
|  70|      b, d|  4|
+----+----------+---+

Использование pyspark Я использовал следующий код для получения вышеуказанного результата.Я не мог переписать такой код через scala.Есть ли у вас предложения?Заранее благодарю, желаю вам хорошего дня.

код pyspark:

# dataframe
l = [("66", "a,b,c,a", "4"),("67", "a,f,g,t", "0"),("70", "b,b,b,d", "4")]
#spark.createDataFrame(l).show()
df1 = spark.createDataFrame(l, ['KEY1', 'KEY2','ID'])


# function
import re
import numpy as np
# drop duplicates in a row
def drop_duplicates(row):
    # split string by ', ', drop duplicates and join back
    words = re.split(',',row)
    return ', '.join(np.unique(words))


# drop duplicates
from pyspark.sql.functions import udf

drop_duplicates_udf = udf(drop_duplicates)

dataset2 = df1.withColumn('KEY2', drop_duplicates_udf(df1.KEY2))
dataset2.show()

Answer 1

import org.apache.spark.sql._

 val dfUpdated = dataset1.rdd.map{
     case Row(x: String, y: String,z:String) => (x,y.split(",").distinct.mkString(", "),z)
 }.toDF(dataset1.columns:_*)

В искровом корпусе:

scala> val dataset1 = Seq(("66", "a,b,c,a", "4"), ("67", "a,f,g,t", "0"), ("70", "b,b,b,d", "4")).toDF("KEY1", "KEY2", "ID")    
dataset1: org.apache.spark.sql.DataFrame = [KEY1: string, KEY2: string ... 1 more field]

scala> dataset1.show
+----+-------+---+
|KEY1|   KEY2| ID|
+----+-------+---+
|  66|a,b,c,a|  4|
|  67|a,f,g,t|  0|
|  70|b,b,b,d|  4|
+----+-------+---+

scala> val dfUpdated = dataset1.rdd.map{
           case Row(x: String, y: String,z:String) => (x,y.split(",").distinct.mkString(", "),z)
       }.toDF(dataset1.columns:_*)
dfUpdated: org.apache.spark.sql.DataFrame = [KEY1: string, KEY2: string ... 1 more field]

scala> dfUpdated.show
+----+----------+---+
|KEY1|      KEY2| ID|
+----+----------+---+
|  66|   a, b, c|  4|
|  67|a, f, g, t|  0|
|  70|      b, d|  4|
+----+----------+---+

Answer 2

Dataframe решение

scala> val df = Seq(("66", "a,b,c,a", "4"), ("67", "a,f,g,t", "0"), ("70", "b,b,b,d", "4")).toDF("KEY1", "KEY2", "ID")
df: org.apache.spark.sql.DataFrame = [KEY1: string, KEY2: string ... 1 more field]

scala> val distinct :String => String = _.split(",").toSet.mkString(",")
distinct: String => String = <function1>

scala> val distinct_id = udf (distinct)
distinct_id: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

scala> df.select('key1,distinct_id('key2).as("distinct"),'id).show
+----+--------+---+
|key1|distinct| id|
+----+--------+---+
|  66|   a,b,c|  4|
|  67| a,f,g,t|  0|
|  70|     b,d|  4|
+----+--------+---+


scala>

Answer 3

Возможно, существует более оптимизированное решение, но оно может вам помочь.

val rdd2 = dataset1.rdd.map(x => x(1).toString.split(",").distinct.mkString(", "))

//, а затем преобразовать его в набор данных // или

val distinctUDF = spark.udf.register("distinctUDF", (s: String) => s.split(",").distinct.mkString(", "))

dataset1.createTempView("dataset1")

spark.sql("Select KEY1, distinctUDF(KEY2), ID from dataset1").show