Spark SQLでHive MapJoinを実現
2366 ワード
転載先:
http://lxw1234.com/archives/2015/06/296.htm
Hiveでは、大きなテーブルと小さなテーブルでjoinを行う場合、Hiveは自動または手動でMapJoinを使用して、小さなテーブルのデータをDistributeCacheにロードし、MapTaskを使用して大きなテーブルをスキャンすると同時にjoinを完了することができ、joinのパフォーマンスが非常に向上します.
SparkSQLでは、自動または手動でのMapJoinの使用は現在サポートされていません.融通のきく方法は、小さな時計をcacheしてから、大きな時計とjoinをします.SparkSQLのcacheの役割は、各Workerのメモリに小表データをブロードキャストすることと、DistributeCacheにロードすることです.
具体的には、次のようになります.
create table t_lxw1234 as
SELECT a.cookieid,
b.brand,
a.orderid AS ad_activity_id,
a.orderitemid AS ad_id,
a.siteid AS media_id,
a.inventoryid AS ad_area_id,
SUM(1) AS pv
FROM lxw1234.t_log a
join lxw1234.t_config b
ON (a.orderid = b.ad_activity_id)
WHERE a.pt = '2015-06-15'
GROUP BY a.cookieid,
b.brand,
a.orderid,
a.orderitemid,
a.siteid,
a.inventoryid
上のSQLの中で、大きい表lxw 1234.t_logには3億件以上の記録があり、小表lxw 1234.t_configには1000件以上しかありませんが、一般的にSparkSQLの実行計画は下図のようになります.
まず2枚のテーブルを別々にスキャンし、その後ShuffledHashJoinを行うが、このステップでは、小さなテーブルのデータ量が非常に小さい、すなわち関連できるキー値が少ないため、ここでデータの傾きが発生し、最後のいくつかのtask処理のデータ量が非常に大きくなり、メモリがオーバーフローしてエラーが報告されるまで、図のように:
SparkSQLにはCACHE TABLEのコマンドが用意されています.テーブルまたはクエリーをブロードキャストできます.コマンドは次のとおりです.
CACHE TABLE t_config AS SELECT ad_activity_id,brand FROM lxw1234.t_config
これは、t_configこのtableはDistributeCacheにロードされ、次にこのメモリテーブルと大きなテーブルでjoinをします.
create table t_lxw1234 as
SELECT a.cookieid,
b.brand,
a.orderid AS ad_activity_id,
a.orderitemid AS ad_id,
a.siteid AS media_id,
a.inventoryid AS ad_area_id,
SUM(1) AS pv
FROM lxw1234.t_log a
join t_config b
ON (a.orderid = b.ad_activity_id)
WHERE a.pt = '2015-06-15'
GROUP BY a.cookieid,
b.brand,
a.orderid,
a.orderitemid,
a.siteid,
a.inventoryid
実行計画を参照してください.
今回は、一つのステージで、大きなテーブルのスキャンと小さなテーブルとのBroadcastHashJoinを完成させ、性能的にはもちろん、すぐに走り終えました.
Hiveで同じ文を試してみましたが、HiveではMapJoinを歩き、SparkSQLより50%近くの時間を使用していますが、HiveではMapReduceが消費するリソースに注意してください.
しかし、SparkSQLはリソースの何倍も消費しており、HDFSからデータを読むにもかかわらず、SparkはMapReduceよりも優れていることが確認されています.
また、SparkSQLはHiveテーブル(HDFS)からデータを読み、全て用のNODE_LOCAL taskは、ANYであれば少し遅くなり、大きなネットワークリソースを消費します.
転載は明記してください:lxwの大きいデータの畑 » Spark SQLでHive MapJoinを実現
http://lxw1234.com/archives/2015/06/296.htm
Hiveでは、大きなテーブルと小さなテーブルでjoinを行う場合、Hiveは自動または手動でMapJoinを使用して、小さなテーブルのデータをDistributeCacheにロードし、MapTaskを使用して大きなテーブルをスキャンすると同時にjoinを完了することができ、joinのパフォーマンスが非常に向上します.
SparkSQLでは、自動または手動でのMapJoinの使用は現在サポートされていません.融通のきく方法は、小さな時計をcacheしてから、大きな時計とjoinをします.SparkSQLのcacheの役割は、各Workerのメモリに小表データをブロードキャストすることと、DistributeCacheにロードすることです.
具体的には、次のようになります.
create table t_lxw1234 as
SELECT a.cookieid,
b.brand,
a.orderid AS ad_activity_id,
a.orderitemid AS ad_id,
a.siteid AS media_id,
a.inventoryid AS ad_area_id,
SUM(1) AS pv
FROM lxw1234.t_log a
join lxw1234.t_config b
ON (a.orderid = b.ad_activity_id)
WHERE a.pt = '2015-06-15'
GROUP BY a.cookieid,
b.brand,
a.orderid,
a.orderitemid,
a.siteid,
a.inventoryid
上のSQLの中で、大きい表lxw 1234.t_logには3億件以上の記録があり、小表lxw 1234.t_configには1000件以上しかありませんが、一般的にSparkSQLの実行計画は下図のようになります.
まず2枚のテーブルを別々にスキャンし、その後ShuffledHashJoinを行うが、このステップでは、小さなテーブルのデータ量が非常に小さい、すなわち関連できるキー値が少ないため、ここでデータの傾きが発生し、最後のいくつかのtask処理のデータ量が非常に大きくなり、メモリがオーバーフローしてエラーが報告されるまで、図のように:
SparkSQLにはCACHE TABLEのコマンドが用意されています.テーブルまたはクエリーをブロードキャストできます.コマンドは次のとおりです.
CACHE TABLE t_config AS SELECT ad_activity_id,brand FROM lxw1234.t_config
これは、t_configこのtableはDistributeCacheにロードされ、次にこのメモリテーブルと大きなテーブルでjoinをします.
create table t_lxw1234 as
SELECT a.cookieid,
b.brand,
a.orderid AS ad_activity_id,
a.orderitemid AS ad_id,
a.siteid AS media_id,
a.inventoryid AS ad_area_id,
SUM(1) AS pv
FROM lxw1234.t_log a
join t_config b
ON (a.orderid = b.ad_activity_id)
WHERE a.pt = '2015-06-15'
GROUP BY a.cookieid,
b.brand,
a.orderid,
a.orderitemid,
a.siteid,
a.inventoryid
実行計画を参照してください.
今回は、一つのステージで、大きなテーブルのスキャンと小さなテーブルとのBroadcastHashJoinを完成させ、性能的にはもちろん、すぐに走り終えました.
Hiveで同じ文を試してみましたが、HiveではMapJoinを歩き、SparkSQLより50%近くの時間を使用していますが、HiveではMapReduceが消費するリソースに注意してください.
しかし、SparkSQLはリソースの何倍も消費しており、HDFSからデータを読むにもかかわらず、SparkはMapReduceよりも優れていることが確認されています.
また、SparkSQLはHiveテーブル(HDFS)からデータを読み、全て用のNODE_LOCAL taskは、ANYであれば少し遅くなり、大きなネットワークリソースを消費します.
転載は明記してください:lxwの大きいデータの畑 » Spark SQLでHive MapJoinを実現