Sparkパフォーマンスの最適化ブロードキャスト変数の使用方法とKryoシーケンス化
3029 ワード
ブロードキャスト変数
Sparkアプリケーションでは、共有変数がよく使用されます.よく知られているように、Sparkは並列計算フレームワークであり、この変数に対しては、executorのtaskがアクセスするたびにコピーをコピーして使用します.次の図に示します.
このデフォルト方式では、システムのメモリが極めて大きくなります.クラスタに1024個のtaskがあると仮定することができます.この共有変数のサイズは1 Mと仮定すると、クラスタに1024部コピーされ、1個のGのデータがネットワークに転送され、これらのコピーに空間を割り当てるのに1 Gのメモリが必要になります.これはシステムにどのような影響がありますか.
システムメモリが不足し、RDDが永続化されていない場合、メモリに永続化できないため、ディスクに永続化する必要がある場合、後続の操作は頻繁なディスクIOによって速度が遅くなり、パフォーマンスが低下します.taskでオブジェクトを作成するとスタック内にメモリが不足していることが判明し,GC操作が必要となり,GCを行うとワークスレッドが一時停止し,メモリが深刻に不足している場合,頻繁なGCがSparkジョブの速度に与える影響が考えられる.以上の問題に対して、パフォーマンスの最適化はどのようにすればいいのでしょうか.
この場合,Broadcastを用いて,このtaskごとに必要な共有変数をブロードキャストすることができる.
上の図から、taskごとにこの変数を使用する必要がある場合、1部がコピーされることがわかります.ブロードキャスト変数を使用すると、まずブロードキャスト変数がDriverにコピーされ、各executorのtaskがその変数を使用すると、まず各executorのBlockManagerにその変数のコピーがあるかどうかをチェックし、ない場合はDriverにコピーをBlockManagerにコピーし、そのexecutorの各taskに使用し、次のexecutorのtaskにこの変数を使用する必要がある場合、そのBlockManagerは、Driverからコピーをコピーしたり、近いexecutorのBlockManagerからコピーしたりすることができます.(各executorのBlockManagerの役割は、各executorに対応するメモリとディスクのデータを管理することです.)その原理図は上記の通りです.
デフォルトでは、1024個のtaskで1 Gメモリを消費する必要がありますが、これらのtaskを分割するために50個のexecutorがあれば、50個のコピーだけで50 Mメモリを消費することができ、メモリの消費量を約20倍に節約できます.また、レプリカのコピーをDriverからコピーする必要がなく、他のexecutorからコピーする場合、ネットワーク転送によるパフォーマンス消費も少なく、ブロードキャスト変数を使用するとメモリを大幅に節約でき、パフォーマンスが大幅に向上することが予想されます.
ブロードキャスト変数はどのように使用しますか?
たとえば、共有変数はmapタイプの変数であり、Sparkコンテキストを使用してブロードキャスト変数を作成できます.
Broadcast> broadcast=sc.broadcast(map);
taskで使用する場合、valueメソッドまたはgetValueメソッドを使用して値を取得できます.
Map map=broadcast.value;
Kryoシーケンス化
ブロードキャスト変数によってネットワークの伝送圧力を低減し、メモリを大幅に節約した後、さらに最適化することができます.シーケンス化は、データの消費メモリサイズを削減するのに良い選択です.
デフォルトのSparkではjavaシーケンス化メカニズムが使用されます.つまり、ObjectOutputStream/ObjectInputStream、オブジェクト入出力ストリームメカニズムによってシーケンス化されます.
デフォルトのこのシーケンス化メカニズムを使用すると、簡単で便利で、自分で構成する必要がなく、シーケンス化が必要なクラスでSerializableインタフェースを実現する必要があります.欠点は,シーケンス化の効率が高くなく,シーケンス化の速度が低く,シーケンス化後のデータ占有空間が依然として大きいことである.したがって,必要に応じてシーケンス化方式を手動で指定して最適化することができる.
SparkはKryoシーケンス化メカニズムをサポートし、Kryoシーケンス化メカニズムは、デフォルトのJavaシーケンス化メカニズムよりも速度が速く、シーケンス化後のデータがより小さく、Javaシーケンス化メカニズムの1/10程度である.
したがって,Kryoシーケンス化機構を用いることで,ネットワークで伝送されるデータをより小さくすることができ,クラスタで消費されるメモリも大幅に減少する.
Kryoシーケンス化作用の場所
Kryoシーケンス化が有効になると、次のいくつかの場所で有効になります.
演算子関数で使用される外部変数.演算子関数で使用する外部変数は、Kryoを使用した後、ネットワーク伝送の性能を最適化し、クラスタ内のメモリの占有と永続化RDDの消費を最適化する際にシーケンス化することができる、例えばStorageLevel.MEMORY_ONLY_SER.RDDを永続化し、メモリの使用量と消費量を最適化する.永続化RDDが消費するメモリが少ないほどtaskが実行されるときに作成されるオブジェクトは,頻繁にメモリをフルにすることなくGCが頻繁に発生する.shuffleでシーケンス化すると、ネットワーク転送のパフォーマンスが最適化されます.Kryoシーケンス化の使用方法
ステップ1:SparkConfにシーケンス化属性sparkを設定.serializer、値org.apache.spark.serializer.KryoSerializer.
ステップ2:Kryoシーケンス化を使用するカスタムクラスを登録し、SparkConfを使用します.registerKryoClasses()メソッドを登録します.
例:
ブロードキャスト変数とシーケンス化を使用するのは簡単ですが、システムの性能に与える影響を無視することはできません.小さな修正でシステムの性能を何倍も向上させることができるかもしれません.これも私たちの性能最適化の重点で、すべての細部を無視することはできません.
Sparkアプリケーションでは、共有変数がよく使用されます.よく知られているように、Sparkは並列計算フレームワークであり、この変数に対しては、executorのtaskがアクセスするたびにコピーをコピーして使用します.次の図に示します.
このデフォルト方式では、システムのメモリが極めて大きくなります.クラスタに1024個のtaskがあると仮定することができます.この共有変数のサイズは1 Mと仮定すると、クラスタに1024部コピーされ、1個のGのデータがネットワークに転送され、これらのコピーに空間を割り当てるのに1 Gのメモリが必要になります.これはシステムにどのような影響がありますか.
システムメモリが不足し、RDDが永続化されていない場合、メモリに永続化できないため、ディスクに永続化する必要がある場合、後続の操作は頻繁なディスクIOによって速度が遅くなり、パフォーマンスが低下します.taskでオブジェクトを作成するとスタック内にメモリが不足していることが判明し,GC操作が必要となり,GCを行うとワークスレッドが一時停止し,メモリが深刻に不足している場合,頻繁なGCがSparkジョブの速度に与える影響が考えられる.以上の問題に対して、パフォーマンスの最適化はどのようにすればいいのでしょうか.
この場合,Broadcastを用いて,このtaskごとに必要な共有変数をブロードキャストすることができる.
上の図から、taskごとにこの変数を使用する必要がある場合、1部がコピーされることがわかります.ブロードキャスト変数を使用すると、まずブロードキャスト変数がDriverにコピーされ、各executorのtaskがその変数を使用すると、まず各executorのBlockManagerにその変数のコピーがあるかどうかをチェックし、ない場合はDriverにコピーをBlockManagerにコピーし、そのexecutorの各taskに使用し、次のexecutorのtaskにこの変数を使用する必要がある場合、そのBlockManagerは、Driverからコピーをコピーしたり、近いexecutorのBlockManagerからコピーしたりすることができます.(各executorのBlockManagerの役割は、各executorに対応するメモリとディスクのデータを管理することです.)その原理図は上記の通りです.
デフォルトでは、1024個のtaskで1 Gメモリを消費する必要がありますが、これらのtaskを分割するために50個のexecutorがあれば、50個のコピーだけで50 Mメモリを消費することができ、メモリの消費量を約20倍に節約できます.また、レプリカのコピーをDriverからコピーする必要がなく、他のexecutorからコピーする場合、ネットワーク転送によるパフォーマンス消費も少なく、ブロードキャスト変数を使用するとメモリを大幅に節約でき、パフォーマンスが大幅に向上することが予想されます.
ブロードキャスト変数はどのように使用しますか?
たとえば、共有変数はmapタイプの変数であり、Sparkコンテキストを使用してブロードキャスト変数を作成できます.
Broadcast> broadcast=sc.broadcast(map);
taskで使用する場合、valueメソッドまたはgetValueメソッドを使用して値を取得できます.
Map map=broadcast.value;
Kryoシーケンス化
ブロードキャスト変数によってネットワークの伝送圧力を低減し、メモリを大幅に節約した後、さらに最適化することができます.シーケンス化は、データの消費メモリサイズを削減するのに良い選択です.
デフォルトのSparkではjavaシーケンス化メカニズムが使用されます.つまり、ObjectOutputStream/ObjectInputStream、オブジェクト入出力ストリームメカニズムによってシーケンス化されます.
デフォルトのこのシーケンス化メカニズムを使用すると、簡単で便利で、自分で構成する必要がなく、シーケンス化が必要なクラスでSerializableインタフェースを実現する必要があります.欠点は,シーケンス化の効率が高くなく,シーケンス化の速度が低く,シーケンス化後のデータ占有空間が依然として大きいことである.したがって,必要に応じてシーケンス化方式を手動で指定して最適化することができる.
SparkはKryoシーケンス化メカニズムをサポートし、Kryoシーケンス化メカニズムは、デフォルトのJavaシーケンス化メカニズムよりも速度が速く、シーケンス化後のデータがより小さく、Javaシーケンス化メカニズムの1/10程度である.
したがって,Kryoシーケンス化機構を用いることで,ネットワークで伝送されるデータをより小さくすることができ,クラスタで消費されるメモリも大幅に減少する.
Kryoシーケンス化作用の場所
Kryoシーケンス化が有効になると、次のいくつかの場所で有効になります.
演算子関数で使用される外部変数.演算子関数で使用する外部変数は、Kryoを使用した後、ネットワーク伝送の性能を最適化し、クラスタ内のメモリの占有と永続化RDDの消費を最適化する際にシーケンス化することができる、例えばStorageLevel.MEMORY_ONLY_SER.RDDを永続化し、メモリの使用量と消費量を最適化する.永続化RDDが消費するメモリが少ないほどtaskが実行されるときに作成されるオブジェクトは,頻繁にメモリをフルにすることなくGCが頻繁に発生する.shuffleでシーケンス化すると、ネットワーク転送のパフォーマンスが最適化されます.Kryoシーケンス化の使用方法
ステップ1:SparkConfにシーケンス化属性sparkを設定.serializer、値org.apache.spark.serializer.KryoSerializer.
ステップ2:Kryoシーケンス化を使用するカスタムクラスを登録し、SparkConfを使用します.registerKryoClasses()メソッドを登録します.
例:
new SparkConf()
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(new Class[]{CategorySortKey.class})
ブロードキャスト変数とシーケンス化を使用するのは簡単ですが、システムの性能に与える影響を無視することはできません.小さな修正でシステムの性能を何倍も向上させることができるかもしれません.これも私たちの性能最適化の重点で、すべての細部を無視することはできません.