如何在Spark2中对百万条记录(每日增量加载)的文件生成序列

Question

我有一个使用scala 2.11.8在spark 2.0中对每日增量表或文件生成代理键的业务场景。我知道"zipwithindex“、"row_num”和"monotonically_increasing_id()“，但它们都不适用于每日增量加载，因为对于今天的加载，我的序列将是1+昨天的序列。累加器也不会工作，因为它是只写的。

例如。场景:直到昨天的加载，我有最后的customer_sk为1001，现在在今天的加载中，我想设置从1002开始到文件结束的customer_sk。

注意:我将有数百万行，程序将在多个节点上并行运行。

提前感谢

Answer 1

1)从表中获取最大customer_sk。

2)然后，当使用row_num时，添加这个最大customer_sk数，这样您的序列就可以从这个数继续。

如果还使用rdd，请将先前的最大值添加到(zipwithindex +1)。

Answer 2

对于那些仍然在寻找示例代码答案的人。

hdfs dfs -cat /user/shahabhi/test_file_2.csv

abhishek，shah,123，pune,2018-12-31,2018-11-30 abhishek，shah,123，pune,2018-12-31,2018-11-30 ravi，sharma,464，孟买，20181231,20181130 Mitesh，shah,987，satara,2018-12-31,2018-11-30 shalabh，nagar,981，satara,2018-12-31,2018-11-30 Gaurav，mehta,235，ujjain，2018/12-31,2018/11/30 Gaurav，mehta,235，ujjain,2018-12-31,2018-11-30 vikas，khanna,123，ujjain,2018-12-31,2018-11-30 vinayak，kale，789，浦那，2018-12-31,2018-11-30

火花码

import org.apache.spark.sql.functions.monotonically_increasing_id

val df =spark.read.csv("/user/shahabhi/test_file_2.csv").toDF("name","lname","d_code","city","esd","eed")

df.show()

+--------+------+------+------+----------+----------+ |名称|esd城市| esd| eed| +--------+------+------+------+----------+---------- | lname|d_code|+ |abhishek| shah| 123| pune|2018-12-31|2018-11-30| |abhishek| shah| 123| pune|2018-12-31|2018-11-30| |拉维|sharma|464|孟买| 20181231| 20181130| | Mitesh| shah| 987|satara|2018-12-31|2018-11-30| | shalabh| nagar| 981|satara|2018-12-31|2018-11-30| | Gaurav| mehta| 235|2018/11/30| Gaurav| mehta| 235|ujjain|2018-12-31|2018-11-30| | vikas|khanna| 123|ujjain|2018-11-31| vinayak| 789| pune|12-31|2018-11-30|--+------+------+------+----------+----------+

val df_2=df.withColumn("surrogate_key", monotonically_increasing_id())

df_2.show()

esd|名称| lname|d_code|城市|esd| eed|surrogate_key| +-+ |abhishek| shah| 123| pune|2018-12-31|2018-11-30| 0| |abhishek| shah| 123| pune|2018-12-31|2018-11-30| 1| | ravi|sharma| 464|孟买| 20181231| 20181130| 2| Mitesh| shah| 987|satara|2018-12-31|2018-11-30| 3| | shalabh| nagar| 981|satara|2018-12-31|2018-11-30| 4|| Gaurav| mehta| 235|ujjain|2018/12/31|2018/11/30| 5| Gaurav| mehta| 235|ujjain|2018-12-31|2018-11-30| 6| vikas|khanna| 123|ujjain|2018-12-31|2018-11-30| 7| vinayak| kale| 789| pune|2018-12-31|2018-11-30| 8| +-+

val df_3=df.withColumn("surrogate_key", monotonically_increasing_id()+1000)
df_3.show()

esd|名称| lname|d_code|城市|esd| eed|surrogate_key| +-+-+ |abhishek| shah| 123| pune|2018-12-31|2018-11-30| 1000 | abhishek| shah| 123| pune|2018-12-31|2018-11-30| 1001| | ravi|sharma| 464|孟买| 20181231| 20181130| 1002| | Mitesh| shah| 987|satara|2018-12-31| nagar| 981|satara|2018-12-31|2018-11-30| 1004| | Gaurav| mehta| 235|ujjain|2018年/12/31|2018/11/30| 1005| |高拉夫|梅塔|235|乌贾因|2018-12-31|2018-11-30| 1006| |维卡斯|卡纳|123|乌贾因|2018-12-31|2018-11-30| 1007| |维纳亚克|甘蓝| 789|普纳|2018-12-31|2018-11-30| 1008| +--------+------+------+------+----------++