火花增量加载覆盖旧记录

Question

我需要使用Spark (PySpark)对表进行增量加载

下面是一个例子：

第一天

id | value
-----------
1  | abc
2  | def

第2天

id | value
-----------
2  | cde
3  | xyz

预期结果

id | value
-----------
1  | abc
2  | cde
3  | xyz

这可以很容易地在关系数据库中完成，

想知道这是否可以在火花或其他转型工具，例如普雷斯托？

Answer 1

这就是你要的!第一个Dataframe：

 >>> list1 = [(1, 'abc'),(2,'def')]
 >>> olddf = spark.createDataFrame(list1, ['id', 'value'])
 >>> olddf.show();
 +---+-----+
 | id|value|
 +---+-----+
 |  1|  abc|
 |  2|  def|
 +---+-----+

第二个Dataframe：

>>> list2 = [(2, 'cde'),(3,'xyz')]
>>> newdf = spark.createDataFrame(list2, ['id', 'value'])
>>> newdf.show();
+---+-----+
| id|value|
+---+-----+
|  2|  cde|
|  3|  xyz|
+---+-----+

现在，使用完全外部连接和合并来连接和合并这两个数据文件，同时使用合并函数，同时使用用户定义的值来选择和替换空值。

from pyspark.sql.functions import *

>>> df = olddf.join(newdf, olddf.id == newdf.id,'full_outer').select(coalesce(olddf.id,newdf.id).alias("id"),coalesce(newdf.value,olddf.value).alias("value"))
>>> df.show();
+---+-----+
| id|value|
+---+-----+
|  1|  abc|
|  3|  xyz|
|  2|  cde|
+---+-----+

我希望这能解决你的问题。:-)

Answer 2

数据附加是通过union函数实现的。我将演示一个例子，并创建2个数据，正如您在问题中提到的。

from pyspark.sql.types import Row
df1 = sqlContext.createDataFrame([Row(id=1,value="abc"),Row(id=2,value="def")])

df1.show()
+---+-----+
| id|value|
+---+-----+
|  1|  abc|
|  2|  def|
+---+-----+

df2 = sqlContext.createDataFrame([Row(id=2,value="cde"),Row(id=3,value="xyz")])
df2.show()
+---+-----+
| id|value|
+---+-----+
|  2|  cde|
|  3|  xyz|
+---+-----+

让我们在这两个数据文件之间执行一个union，您将得到所需的结果。

df2.union(df1).dropDuplicates(["id"]).show()
+---+-----+
| id|value|
+---+-----+
|  1|  abc|
|  3|  xyz|
|  2|  cde|
+---+-----+

可以使用asc从pyspark.sql.functions对输出进行排序。

from pyspark.sql.functions import asc


df2.union(df1).dropDuplicates(["id"]).sort(asc("id")).show()
+---+-----+
| id|value|
+---+-----+
|  1|  abc|
|  2|  cde|
|  3|  xyz|
+---+-----+

Answer 3

解决方法是，在dataframe中添加一个日期列，然后根据id排序，然后在降序中按日期顺序排序，然后取== 1，它将始终根据id给出最新的记录。

df.("rank", rank().over(Window.partitionBy($"id").orderBy($"date".desc)))
  .filter($"rank" === 1)
  .drop($"rank")
  .orderBy($"id")
  .show