pyspark方法只获取更新和新记录

3phpmpom 于 2021-06-01 发布在 Hadoop

关注(0)|答案(1)|浏览(351)

我使用的是pyspark2.1，下面是我的Dataframe
昨天数据
1，纳格拉吉，凯沙夫，2017-11-20 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038
2，raghu，人力资源，2017-11-20 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038
今天的数据
1，纳格拉吉，k，2017-11-21 00:02:39.867000000 2017-11-21 00:02:39.867000000
2，拉格胡，人力资源，2017-11-21 00:02:39.867000000 2017-11-20 00:02:39.867000000
3，ramya，govindaraju，2017-11-21 00:02:39.867000000 2017-11-20 00:02:39.867000000
我的输出
1，纳格拉吉，k，2017-11-21 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038
3，ramya，govindaraju，2017-11-21 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038
我不应该得到在两个Dataframe中都存在的记录，因为名称中只有第一个记录发生了变化，我应该得到这个记录，记录编号3是新记录。
我用了下面的逻辑

df =today_data_df.select("id").subtract(yesterdata_data_df.select("id")).toDF('d1').join(today_data_df,col('d1')==today_data_df.id).drop('d1')

输出为：
3，ramya，govindaraju，2017-11-21 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038
但我应该得到下面给出的请帮助
1，纳格拉吉，k，2017-11-21 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038 3，拉米耶，戈文达拉朱，2017-11-21 00:02:39.867000002017-11-20 00:02:39.8670000020171120060038

hadoop python pyspark

来源：https://stackoverflow.com/questions/47426943/pyspark-method-to-get-only-update-and-new-records

1条答案

按热度按时间

kyxcudwk1#

我假设有一个名称字段包含'，'

ydata=[(1,'Nagraj,Keshav','2017-11-20 00:02:39.867000000','2017-11-20 00:02:39.867000000',20171120060038),(2,'Raghu,HR','2017-11-20 00:02:39.867000000','2017-11-20 00:02:39.867000000',20171120060038)]
yschema=['id','name','fdate','tdate','stamp']
tdata=[(1,'Nagraj,K','2017-11-21 00:02:39.867000000','2017-11-21   00:02:39.867000000',20171120060038),(2,'Raghu,HR','2017-11-21 00:02:39.867000000','2017-11-20 00:02:39.867000000',20171120060038),(3,'Ramya,Govindaraju','2017-11-21 00:02:39.867000000','2017-11-20 00:02:39.867000000',20171120060038)]
ydf=spark.createDataFrame(ydata,yschema)
tdf=spark.createDataFrame(tdata,yschema)
newdf=tdf.select('id','name').subtract(ydf.select('id','name'))

newdf.join(tdf,newdf['id']==tdf['id']).drop(tdf['id']).drop(tdf['name']).show()

输出：

| id|             name|               fdate|               tdate|           stamp|
    +---+-----------------+--------------------+--------------------+--------------+
    |  1|         Nagraj,K|2017-11-21 00:02:...|2017-11-21   00:02:...|20171120060038|
    |  3|Ramya,Govindaraju|2017-11-21 00:02:...|2017-11-20 00:02:...|20171120060038|

赞(0）回复(0）举报 2021-06-01

我来回答

pyspark方法只获取更新和新记录

1条答案

相关问题

热门标签

最新问答