PySparkで大きなテーブル結合を最適化する

Question

私は1日におよそ500M行の大きなファクトテーブルを持っています。テーブルはregion_dateによって分割されます。

私は毎日6ヵ月間のデータをスキャンしなければならず、id &の日付列に基づいて別の小さなサブセット（1M行）と外部結合を残して、2つの集計値を計算します：id（右側）

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 

最適化し、配布/データを分割するための最良の方法は何ですか？&合計（事実）

私SparkSQLは、このようになりますかクエリは3時間以上実行されます。試しましたspark.sql.shuffle.partitions = 700

「id」レベルで日次データをロールアップすると、1日あたり約5M行になります。データを最初にロールアップしてから結合する必要がありますか？

ありがとう、

Ram。

Answer 1

クエリにいくつかのフィルタ条件があるため、データ量を減らすためにクエリを2つのクエリに分割できると考えました。あなたが実行計画を含めることができ

 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テーブル