Pyspark Join 使用每月范围

Posted 2023-04-15

【中文标题】Pyspark Join 使用每月范围

【英文标题】：Pyspark Joing using monthly range

【发布时间】：2019-05-22 17:37:50

【问题描述】：

Hive 中有三个表 A , B 表具有以下列，并根据天进行分区。我们需要提取从 2016 年 1 月 1 日到 2016 年 12 月 31 日的数据。我刚刚提到了样本，但这些记录在 1 年内以百万为单位。

ID Day Name Description
1   2016-09-01  Sam   Retail
2   2016-01-28  Chris Retail
3   2016-02-06  ChrisTY Retail
4   2016-02-26  Christa Retail
3   2016-12-06  ChrisTu Retail
4   2016-12-31  Christi Retail

表 B

ID SkEY
1  1.1
2  1.2
3  1.3

表 C

Start_Date  End_Date Month_No
2016-01-01 2016-01-31 1
2016-02-01 2016-02-28 2
2016-03-01 2016-03-31 3
2016-04-01 2016-04-30 4
2016-05-01 2016-05-31 5
2016-06-01 2016-06-30 6
2016-07-01 2016-07-31 7
2016-08-01 2016-08-31 8
2016-09-01 2016-09-30 9
2016-10-01 2016-10-30 10
2016-11-01 2016-11-31 11
2016-12-01 2016-12-31 12

我尝试在 spark 中编写代码，但没有成功，导致连接时出现 cartis 产品，性能也很差

Df_A=spark.sql("select * from A join B where a.day>=b.start_date
     and a.day<=b.end_date and b.month_no=(I)") 

实际输出应该有 pyspark 中的代码，其中 A 加入 B，每个月都需要处理。 I 的值应与月份日期一起自动从 1 增加到 12。 A Join B 如上图 A Join C using ID 以及性能应该不错

【参考方案1】：

from pyspark.sql import sparksession
from pyspark.sql import functions as F
from pyspark import HiveContext
hiveContext= HiveContext(sc)


def UDF_df(i):
print(i[0])
ABC2=spark.sql("select * From A where day where day    
='0'.format(i[0]))
Join=ABC2.join(Tab2.join(ABC2.ID == Tab2.ID))\
.select(Tab2.skey,ABC2.Day,ABC2.Name,ABC2.Description)
Join\
 .select("Tab2.skey","ABC2.Day","ABC2.Name","ABC2.Description")
 .write\
 .mode("append")\
 .format("parquet')\
 .insertinto("Table")
ABC=spark.sql("select distinct day from A where day<= ' 2016-01-01' and day<='2016-
12-31'")
Tab2=spark.sql("select * from B where day is not null)
for in in ABC.collect():
UDF_df(i)

The following query is working but taking a long time as the number of  
columns are around 60(just used sample 3). Also didn't join Table C as I 
wasn't sure how to join to avoid cartisan join. performance isn't good, am 
not sure how to optimise the query.