Pyspark

Question

我需要在 pyspark 中创建一个等效的业务当前视图，我有一个历史文件和一个增量文件（包含 id 和日期）。我需要创建最终数据框，每个 id 和该记录应该是最新的。

df1=sql_context.createDataFrame([("3000", "2017-04-19"), ("5000", "2017-04-19"), ("9012", "2017-04-19")], ["id", "date"])
df2=sql_context.createDataFrame([("3000", "2017-04-18"), ("5120", "2017-04-18"), ("1012", "2017-04-18")], ["id", "date"])

df3=df2.union(df1).distinct()

+----+----------+
|  id|      date|
+----+----------+
|3000|2017-04-19|
|3000|2017-04-18|
|5120|2017-04-18|
|5000|2017-04-19|
|1012|2017-04-18|
|9012|2017-04-19|

+----+----------+

我尝试做一个 union 并做一个 distinct ，它为我提供了两个日期的 id=3000，因为我只需要为 date=2017-04-19 记录 id=300

甚至减法也不起作用，因为它 returns 任一 df 的所有行。

期望的输出：-

+----+----------+
|  id|      date|
+----+----------+
|3000|2017-04-19|
|
|5120|2017-04-18|
|5000|2017-04-19|
|1012|2017-04-18|
|9012|2017-04-19|
+----+----------+

Answer 1

希望对您有所帮助！

from pyspark.sql.functions import unix_timestamp, col, to_date, max

#sample data
df1=sqlContext.createDataFrame([("3000", "2017-04-19"),
                                ("5000", "2017-04-19"),
                                ("9012", "2017-04-19")],
                               ["id", "date"])
df2=sqlContext.createDataFrame([("3000", "2017-04-18"),
                                ("5120", "2017-04-18"),
                                ("1012", "2017-04-18")],
                               ["id", "date"])
df=df2.union(df1)
df.show()

#convert 'date' column to date type so that latest date can be fetched for an ID
df = df.\
    withColumn('date_inDateFormat',to_date(unix_timestamp(col('date'),"yyyy-MM-dd").cast("timestamp"))).\
    drop('date')

#get latest date for an ID
df = df.groupBy('id').agg(max('date_inDateFormat').alias('date'))
df.show()

输出为：

+----+----------+
|  id|      date|
+----+----------+
|5000|2017-04-19|
|1012|2017-04-18|
|5120|2017-04-18|
|9012|2017-04-19|
|3000|2017-04-19|
+----+----------+

注意：如果回答对您有帮助，请不要忘记let SO know

Pyspark - 在 pyspark 中创建业务当前视图的等效项

Pyspark - Create Equivalent of Business current view in pyspark

pyspark-sql