在PySpark中优化大表连接

Question

我有一个很大的事实表，每天大约有500M行。表按照region_date进行分区。

我必须每天扫描6个月的数据，左外连接与另一个更小的子集(1M行)基于id &日期列，并计算两个聚合值:sum(fact)如果id存在于右表∑()

我的SparkSQL是这样的:

SELECT
    a.region_date,
    SUM(case
          when t4.id is null then 0
          else a.duration_secs
        end) matching_duration_secs
    SUM(a.duration_secs) total_duration_secs
 FROM fact_table a LEFT OUTER JOIN id_lookup t4
       ON a.id = t4.id
      and a.region_date = t4.region_date
 WHERE a.region_date >= CAST(date_format(DATE_ADD(CURRENT_DATE,-180), 'yyyyMMdd') AS BIGINT)
   AND a.is_test = 0
   AND a.desc = 'VIDEO'
GROUP BY a.region_date

优化和分发/分区数据的最佳方法是什么?查询现在运行了3个多小时。我试过spark.sql.shuffle.partitions = 700

如果我在"id"级别上滚动每日数据，则每天大约有5M行。我应该先汇总数据，然后再进行连接吗?

谢谢,

Ram。

Answer 1

因为在你的查询中有一些过滤条件，我认为你可以把你的查询分成两个查询，首先减少数据量。

       table1 = select * from fact_table 
       WHERE a.region_date >= CAST(date_format(DATE_ADD(CURRENT_DATE,-180), 'yyyyMMdd') AS BIGINT)
       AND a.is_test = 0
       AND a.desc = 'VIDEO'

则可以使用比原表小得多的新表来连接id_lookup表