Spark SQL - сравнение данных - PullRequest
Новая
       27

Spark SQL - сравнение данных

0 голосов
/ 23 февраля 2019

Как лучше всего сравнить два файла csv (миллионы строк) с одной и той же схемой со столбцом первичного ключа и распечатать различия.Например,

CSV1 

Id  name   zip   
1   name1  07112  
2   name2  07234  
3   name3  10290

CSV2 

Id  name    zip   
1   name1   07112  
2   name21  07234  
4   name4   10290

Сравнение измененного файла CSV2 с исходными данными CSV1,

Выходные данные должны быть 

Id name    zip   
2  name21  07234 Modified  
3  name3   10290 Deleted  
4  name4   10290 Added

Впервые в Spark SQL я подумываю импортировать данные в таблицы Hive, а затем запустить Spark SQL, чтобы определить изменения.

1) Существует ли какой-либо метод, модифицированный строкой, чтобы определить, была ли строка модифицирована вместо сравнения значений в каждом столбце?2) Есть ли лучший подход для реализации с использованием Spark или других инструментов HDFS?

Ценю обратную связь

1 Ответ

0 голосов
/ 23 февраля 2019

Существует много подходов;это то, что может делать вещи параллельно: 

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

val origDF = sc.parallelize(Seq(
  ("1", "a", "b"),
  ("2", "c", "d"),
  ("3", "e", "f")
)).toDF("k", "v1", "v2")

val newDF = sc.parallelize(Seq(
  ("1", "a", "b"),
  ("2", "c2", "d"),
  ("4", "g", "h")
)).toDF("k", "v1", "v2")

val df1 = origDF.except(newDF) // if k not exists in df2, then deleted
//df1.show(false)
val df2 = newDF.except(origDF) // if k not exists in df1, then added
//df2.show(false)

                           // if no occurrence in both dfs, then the same
                           // if k exists in both, then k in df2 = modified

df1.createOrReplaceTempView("df1")
df2.createOrReplaceTempView("df2")

val df3 = spark.sql("""SELECT df1.k, df1.v1, df1.v2, "deleted" as operation
                         FROM  df1 
                        WHERE NOT EXISTS (SELECT df2.k 
                                            FROM df2
                                            WHERE df2.k = df1.k)
                          UNION
                       SELECT df2.k, df2.v1, df2.v2, "added" as operation
                         FROM  df2 
                         WHERE NOT EXISTS (SELECT df1.k 
                                             FROM df1
                                            WHERE df1.k = df2.k)
                          UNION
                       SELECT df2.k, df2.v1, df2.v2, "modified" as operation
                         FROM  df2 
                        WHERE EXISTS (SELECT df1.k 
                                        FROM df1
                                        WHERE df1.k = df2.k)

                   """)

df3.show(false)

возвращает: 

+---+---+---+---------+
|k  |v1 |v2 |operation|
+---+---+---+---------+
|4  |g  |h  |added    |
|2  |c2 |d  |modified |
|3  |e  |f  |deleted  |
+---+---+---+---------+

Не так сложно, нет стандартной утилиты.

...