Apache Spark-利点、欠点、希望

私は長い間Apache Sparkの印象を表現したかったので、2018年6月26日に発行されたばかりのPivotalの従業員Robert Bennettからのこの記事が目を引きました 。



これは翻訳ではなく、トピックに関する私の印象とコメントです。

Sparkの人気の理由は何ですか？

引用：

Apache Sparkが非常に人気がある理由は簡単にわかります。 インメモリ、分散、および反復計算を行います。これは、機械学習アルゴリズムを使用する場合に特に便利です。 他のツールでは、中間結果をディスクに書き込んでメモリに読み込む必要がある場合があり、反復アルゴリズムの使用が非常に遅くなる可能性があります。

そもそも、これはほとんどの場合完全に真実ではありません。 メモリに？ まあ、はい、Sparkは試行しますが、ここで他のツールについて書かれていることも行われます。 最終的に、メモリ、プロセッサコア、およびネットワークはリソースが限られているため、遅かれ早かれツールは制限に依存します。



ある意味では、Sparkは従来のmap-reduceよりもメモリ内にあることはありません。 何らかの方法で、データは最終的にディスクに保存されるか（とりわけ、エラーをより確実に乗り切ることができ、最初から計算を開始しない）、ネットワークを介して転送されます（シャッフルなどのプロセス）。 プログラマーとしては、突然必要になった場合に中間結果をディスクに保存して保存することを妨げることはほとんどありません。 テラバイトのデータを言う場合、それらをメモリに保存しますか？ 私はそれを疑います。



他のツール（通常は従来のmap-reduceとして理解される）とは異なり、Sparkでは、リソースの最適な使用について少し考えることを許可し、この使用自体をより最適化します。 そして、最終的な速度は、最終的には、むしろプログラムを書く人の手の真っ直ぐさに依存します。



さらに、著者は、彼にとって最も良いと思われるSparkの品質をリストしています。

魅力的なAPIと遅延実行

一般的に、私はこれに同意します。 開発ツールとしてのSparkは、従来のmap-reduceよりもはるかに便利であり、Apache Crunchなどの条件付き「第2」世代のツールよりもいくらか便利です。 また、たとえばHiveよりも多少柔軟性があり、SQL自体に限定されません。



遅延パフォーマンスは常に良いとは限りません。 HiveとDataSetの回路の違いは、すべてのデータがすでに処理されたときではなく、少し早く診断され、すべてが数時間/日ではなく起動時にすべて診断されると言う方が良い場合があります。

簡単な変換

ここで、著者は主にSparkとPython / Pandas構造の間の変換を念頭に置いていました。 私はこれには程遠いので、発言しません。 おそらく、以下でpySparkについて説明します。

簡単な変換

Sparkのもう1つの利点は、「マップ側参加」ブロードキャスト方式です。 この方法は、テーブルの1つが他のテーブルよりも小さく、個々のマシンに完全に収まる場合に、結合を大幅に高速化します。 小さい方はすべてのノードに送信されるため、大きい方のテーブルのデータを移動する必要はありません。 これは、スキューの問題を軽減するのにも役立ちます。 大きなテーブルの結合キーに大きなスキューがある場合、大きなテーブルから少数のノードに大量のデータを送信して、結合を実行し、それらのノードを圧倒しようとします。

pythonの機能はわかりませんが、私たちのエリアでは、マップ側の結合は素手でもCrunshなどのツールでも簡単に行えます。 これには特別な利点はありませんが、多くの人は、たとえばHiveを知っています。 Hadoopマップ側の結合エコシステムにはインデックスが事実上存在しないため、おそらく一般的な結合最適化ツールの1つです。



変換用のAPIは非常に便利ですが、異種のものです。 「古い」RDD APIは、おそらくもう少し柔軟性がありますが、同時に、特に固定構造クラス（Java Beans）のレベルではなく、Rowと柔軟なデータ構造で作業している場合、ミスを犯す範囲が広がります。 この場合、実際のSparkスキームと予想されるSparkスキームの不一致は非常に一般的です。



DataSet APIに関しては、非常に優れていると言えます。 ある程度練習すれば、SQLと同じくらい簡単にすべてを記述し、UDFで補完して柔軟性を高めることができます。 同時に、UDF自体はHiveよりも簡単に記述でき、複雑なデータ構造（配列、マップ、構造）からJavaに戻る場合にのみ、そしておそらくScalaに構造が期待されるため、いくつかの困難が生じます。



Javaポートpymorphy2のようなものをUDFの形で非常に簡単に使用できたとしましょう。 またはジオコーダー。 本質的に、必要なのは、Sparkシリアル化の機能を覚えて、UDFを適切に初期化することだけです。



しかし、一方でSpark ML APIは、まったく異なる人々によって設計されているように見えます。 これは彼が悪いという意味ではありません-彼はただ違うだけです。

オープンソースコミュニティ

Sparkの背後には、大規模なオープンソースコミュニティがあります。 コミュニティはコアソフトウェアを改善し、実用的なアドオンパッケージを提供しています。 たとえば、チームがSpark用の自然言語処理ライブラリを開発しました。 以前は、ユーザーは他のソフトウェアを使用するか、Natural Language ToolkitなどのPythonパッケージを活用するために遅いユーザー定義関数に依存する必要がありました。

ここで一般的に追加するものはありません。 コミュニティは本当に大きく、スキルがあり、フレンドリーです。 Spark用に膨大な数の拡張機能が作成されています。



遅いUDFに関する次の文章は、Pythonistの良心に任せます。Scala/ Java UDFはそれほど遅くはなく、同時に非常に便利です。



自分から追加するもの：

異なる言語での開発

おそらくその人気の理由の1つは、いくつかの開発言語（Scala、Java、Python、およびR）のサポートです。 概して、さまざまな言語のAPIはほぼ同等に便利ですが、このサポートを理想とは呼びません。 Sparkアプリケーションを起動すると、すぐにJava / ScalaとPythonのいずれかを選択でき、一度に言語を組み合わせることはできません。 したがって、pySpark上のアプリケーションの部分（MLまたはNLPの部分が頻繁に記述される）とJava / Scalaの統合は、実際にはファイル/データベースを介してのみ可能です。 まあ、またはカフカ、RESTなどのオプションのようなもの。

ストリーミング

Spark Streaming（完全に異なるHadoop Streamingと混同しないでください）、これはSparkの機能のもう1つの魅力的な部分です。 1つの文で説明すると、これは、たとえば、Kafka、ZeroMQなどからのストリーミングデータの処理です。 データベースから取得したデータと同じ手段で。



すべての魅力は、平均がまったく同じであるという事実に正確にあります。 実際には、Kafkaからのデータの処理を開始するためにプログラム内で何も変更する必要はありません。 map reduce、Crunch、Cascadingのいずれも、このようなトリックを行うことを許可しません。

短所

それぞれに独自の欠点があります（c）。 Sparkを使用するときに直面する問題は何ですか？

クラスター管理

Sparkは、調整と保守が難しいことで有名です。 つまり、最高のパフォーマンスを確保して、データサイエンスの負荷が大きくなっても負荷がかからないようにすることは困難です。 クラスターが適切に管理されていない場合、上記で説明したように、これは「良い」を無効にする可能性があります。 メモリ不足エラーで失敗するジョブは非常に一般的であり、多くの同時ユーザーがいるとリソース管理がさらに難しくなります。

誰かが約束しましたか？ 実際、私はすでに、すべてが素晴らしいものであり、非常に大きなタスクを持たない場合、または必要なだけのリソースを持っている場合など、1つのケースで正確になりうることを既に書きました。つまり、タスクはそれほど複雑ではありません。



最も明白な他のケースでは、Sparkアプリケーションを調整、構成、および保守する必要があります。

固定または動的メモリ割り当てを使用しますか？ Sparkで使用できるクラスターのコアはいくつありますか？ 各エグゼキューターはどのくらいのメモリを取得しますか？ Sparkがデータをシャッフルするときに使用するパーティションはいくつですか？ これらすべての設定をデータサイエンスワークロードに適切に対応させることは困難です。

たとえば、エグゼキュータの数を選択するのは比較的簡単な作業のように思えます。 原則として、データについて何かを知っていれば、この数を安全に計算できます。 しかし、リソースを使用するだけでなく、すべてがより楽しくなります。 プロセスに他のアプリケーションへのアクセスも含まれる場合、...



たとえば、逆ジオコーディング機能を備えたアプリケーションがあります。 また、彼は別のArcGISサーバーに従事しています。 同時に、ArcGISには4つのコアしかなく、Sparkが実行されているHadoopクラスターには多数のノードがあります。そのため、8つのエグゼキューターのみでSparkを選択した場合、ArcGISプロセッサーの負荷曲線は100％にジャンプし、そのまま残ります数時間のアプリケーション操作。 このタスクをSparkに転送すると（アプリケーションコードを以前に書き換えた後）、このタスクにもクラスターリソースを使用できるため、動作時間が数桁短縮されます。



つまり、一定量のリソースが割り当てられるか、これらのリソースが別の方法で管理されるというボトルネックがよくあります（Sparkが影響を与えることはできません）。 したがって、Sparkがこれらのリソースの使用を最適化することを期待するのは単純です。

デバッグ

それが真実です。 ただし、期待されています。 分散並列システムがあり、そのデバッグと監視は重要なタスクです。 SparkUIは監視の問題をある程度解決し、Spark Metricsはパフォーマンス測定を解決しますが、たとえば、デバッガーを使用して実行可能アプリケーションに接続してみます。動作するホストも接続するポートもわかりません。 通常のアプリケーションの場合と同じメトリックスを、たとえばJMXから簡単に取得できます。分散アプリケーションの場合は、ネットワークを介して送信する必要があり、その後でのみ収集できます。 はい、これはすべて比較的悪いです。

PySparkでのUDFパフォーマンスの低下（PySpark UDFの速度低下）

さて、ここで私は何を言うことができますか？ 彼らが戦ったもののために、彼らは何かに出くわしました。 私の知る限り、PythonのUDFは、アプリケーションとUDFの間でデータの二重変換が行われるという事実につながります。 PythonがSparkが実行されるJVMエコシステムの異質な言語であり、UDFが外部で実行されるためです。



ここでアドバイスできるのは1つだけです。Pythonで記述せず、Scala / Javaで記述してください。 このアドバイスが常に望まれているわけではなく、従うことができることは明らかですが、Pythonのバージョンが産業レベルになったときにGraalだけがこの問題をグローバルに解決できるのではないかと思います。

並列処理の最大レベルを保証することは困難です（ハードツーギャランティ最大並列処理）

Sparkの重要な価値提案の1つは分散計算ですが、Sparkが可能な限り計算を並列化することを保証することは困難です。 Sparkは、ジョブのニーズに基づいて、ジョブが使用するエグゼキューターの数を弾性的にスケーリングしようとしますが、多くの場合、それ自体ではスケールアップできません。 そのため、エグゼキューターの最小数を低く設定しすぎると、ジョブは必要なときにそれ以上エグゼキューターを利用できなくなる可能性があります。 また、SparkはRDD（Resilient Distributed Dataset）/ DataFramesをパーティションに分割します。パーティションは、エグゼキューターが実行する最小の作業単位です。 設定するパーティションが少なすぎる場合、すべてのエグゼキューターが作業するのに十分な作業チャンクがない可能性があります。 また、パーティションが少ないとパーティションが大きくなり、エグゼキュータのメモリが不足する可能性があります。

それだけが簡単だったら。 簡単なものから始めましょう-開始のためのパラメーターは、特定のクラスターごとに調整する必要があります。 prodクラスターには、1桁以上のノードがあり、各ノードで何倍ものメモリを使用できます。 Devクラスターの設定は、Prodで起動したときにおそらく過小評価されます。 現在のクラスターの読み込みタスクを考慮し始めると、これはすべてさらに複雑になります。 一般的に、クラスターリソースを割り当てるこのタスクは最適化タスクであり、非常に重要であり、単一の正しい解決策はありません。



パーティションが少ない場合、並列性は不十分です。 また、それらが多すぎる場合、それぞれのサイズは、HDFSブロックのサイズなど、条件付きの下限よりも低くなる可能性があります。 各タスクは起動に費やされるリソースであるため、明らかにオーバーヘッドのコストは生産性よりも速く増加するため、タスクのサイズには下限があり、それ以下に下げる必要はありません。



簡単な例は、かなりの量のディレクトリを必要とするアプリケーションです。 Hadoopでの「通常の」map-reduceタスクの場合、通常、データにコードを配信します。 アプリケーション（Sparkパーツ）をファイル（ファイル）が配置されているクラスターのノードにコピーすると、ディレクトリは既にマップ側の結合に似ているため、コードと一緒に配信する必要があります。 そして突然、各ノードに配信されるデータのサイズが数桁大きくなりました。たとえば、10メガバイト（Spark自体のない小さなSparkアプリケーション）、たとえば20ギガバイト（非常に現実的な場合、アドレス、電話などのデータを正規化するために必要なディレクトリ）そのようなボリュームによってかなり引っ張られる）。 さて、ここにあります-過度の並列処理の代償は明らかです。



おそらく、特定の自然数のパーティションがあります。これは、レプリケーション係数を考慮して、入力ファイルを分割するブロックの数によって決まります。 この数値は、データの読み取りに関して最適に近い可能性があります。 つまり、ファイルに3つのブロックがあり、各ブロックにクラスターの2つのノードにコピーがある場合、6つのスレッドを自然に並列処理し、そのノードで各レプリカを処理できます。 もちろん、Sparkはリソースを動的に割り当てるときにこれらのパラメーターを考慮します。



残念ながら、または幸いなことに、Sparkはクラスターリソースプランナーではありません。 たとえば、糸です。 そのため、Sparkには、すべてのリソースの使用を最適に計画するのに十分な情報がない場合があります。

Hiveとの統合があまり良くない

一方では、SparkはHiveデータとメタデータでうまく機能します。 私が出会ったほとんどのアプリケーションはまさにそれがしていることだと思います。 しかし、迷惑な問題がないわけではありません。 SparkでpartitionByおよびbucketByツールを使用しようとすると、Hiveが作業結果を表示しない可能性が非常に高いと言えます。 さらに、ログのどこかに警告が表示されるだけです。

互換性

残念ながら、このトピックに関する私の経験はかなり悪いです。 Sparkのバージョンが予想と異なるクラスターでアプリケーションを実行しようとすると、複数の問題に遭遇しました。 Spark 2.2.0で開発する場合、2.1および2.3で開始するときに問題がありました。



私たちの場合、何らかの理由でSparkがコーデックの1つ（つまり、snappy）をバージョン2.3で実行しているときに見つけることができなかったとします。 データを書き込む必要がある場合（これは記録時にコーデックを指定し、パックされていないデータを含む任意のものを選択できます）、これはそれほど深刻な問題ではありませんが、急にパックされたものを読む必要がある場合、明らかに運が悪いです。



おそらく問題の一部はインストーラーのエラーが原因でしたが、これはそれほど簡単ではありません。 それにもかかわらず、マイナーバージョン間の移行はよりスムーズになっているはずです。



悲しいかな、Sparkは、同じライン（同じ2.2と2.3）の2つの異なるバージョンの1つのクラスターへのフルタイムの並列インストールを意味しません。

恐ろしいパーティー

APIの厄介さ

Spark APIの多くは非常にエレガントなので、洗練されていない部分が際立っています。 たとえば、配列要素へのアクセスは、Sparkライフのofい部分であると考えています。

配列の操作がそれほどひどいとは言いません。 Spark APIはもともとScalaで作成されていたため、不便な点がいくつかあり、そこには独自のコレクション構造があり、Javaから機能するため、Skalovに減らす必要があります。 したがって、UDFを記述できれば、配列を使って何でもできます。 ええ、はい-Pythonでは、UDFのすべてが悪いです。いつも忘れています。



あまり便利ではなく、あまり効果的でもない-はい、多分。 これは、複雑な構造を扱うための新しい高階関数を導入したSpark 2.4の新しいバージョンを解決しようとしています（これにより、explode / collectの使用が回避されます）。



私の意見では、APIのはるかに不便な側面は、コードを見ると、どの部分がドライバーで実行され、どの部分が他のノードで実行されるかが常に明らかではないということです。 同時に、ノード間でコードを配布するメカニズムには（何らかの方法での）シリアル化が含まれ、エグゼキューターで実行されるコードはシリアル化可能でなければなりません。 シリアル化エラーを理解すると、コードに関する多くの新しい興味深い情報を学ぶことができます:)。

クラスローダー

残念ながら、Sparkコードからアプリケーションコードを分離する問題は十分に解決されていません。 ただし、従来のmap-reduce Hadoopアプリケーションにも同じことが当てはまります。 同時に、HadoopコードはGoogle Guavaなどのライブラリの古いバージョンを使用しますが、他のライブラリはまったく新しいものではありません。 Guavaの作成者が非推奨のメソッドを削除してAPIに後方互換性を導入することを好むことを思い出すと、完全に馬鹿げた画像が得られます-Guavaでコードを新しいバージョンで記述し、実行するとクラッシュします-本当にGuavaバージョンで作業しているからですHadoopから（はるかに古い）、コードが新しいバージョンのメソッドを見つけられないか、新しいバージョンと互換性がないためにHadoopがクラッシュします。 これは非常に典型的な、残念なことに、開発者が2人おきに遭遇する可能性が高い問題です。 Apache Http Componentsライブラリも同様の問題の別の例です。

バインド変数なしのSQL

残念ながら、ペアでクエリを実行するための典型的なコードは次のようになります。



val sqlDF = spark.sql（ "SELECT * FROM people WHERE id = 1"）



APIは、id =？リクエストを実行するオプションを提供しません 実行ごとのパラメータ置換。 さて、SQLインジェクションの問題は作成者を悩ませることはありませんが、開発者はクエリのパラメーターを置き換える必要があります。したがって、特殊文字の置き換えは完全にあなた次第です。 客観性のために、Hiveも同様の問題を抱えています。パラメーターを使用してクエリを定義することもできません。



ただし、おかしなことに、JDBCソースの場合、クエリを記述することさえ正式には不可能です。列のみではなくテーブルのみを指定できます。 非公式には、テーブルの代わりに（dからa、b、cを選択）tのようなものを書くことができますが、これがすべての場合に機能する場合、誰もあなたに確実に伝えません。

成熟度と機能の完全性の欠如

うーん 他人の頭-闇。

もう1つの機能のギャップの例は、Sparkで連続した一意のレコード識別子を作成するのが難しいことです。 連続した一意のインデックス列は、一部のタイプの分析に役立ちます。 ドキュメントによると、「monotonically_increasing_id（）」は各行に一意のIDを生成しますが、IDが連続していることを保証するものではありません。 連続したIDが重要な場合、Sparkの古いRDD形式を使用する必要があります。

私はそのような主張を理解していません。 ソースが利用可能であり、覗いて、少なくともコメントを読むことはかなり可能です：

単調に増加する64ビット整数を返します。

• 生成されたIDは、単調に増加し、一意であることが保証されますが、連続的ではありません。
• 現在の実装では、パーティションIDは上位31ビットに、下位33ビットに配置されます
• 各パーティション内のレコード番号を表します。 仮定は、データフレームが
• 10億未満のパーティション。各パーティションのレコードは80億未満です。

つまり、この関数はパーティション番号を取得し、それにカウンターを追加するだけです。 当然、2回の連続した呼び出しの間に誰もそれを呼び出さないという保証はありません。 1つのSparkアプリケーションは、クラスターの異なるノードで実行されている多くのJVMである可能性があり、おそらく1つのJVM内で実行される多くのスレッドです。



作成者にもう少し考えて、単一の生成ポイントを作成せずに（意図的にボトルネックになります）、ブロックせずに（これは同じになります）、この並列分散システムで必要なIDを正確に生成する方法を考えてみましょうそれ自体で）。

Spark 2.4に期待すること

すでに述べた高階関数

これは本当にいいです。 主なことは働くことです。



実際、これは配列またはマップを操作するための組み込み関数のセットであり、独自の関数（ラムダ）を使用してそれらに対して変換を実行する機能です。



ここで、いくつかの使用例を見ることができます。

新しい実行モード

これは、いわゆるbarierスケジューラおよびランタイムです。 著者はそれを機械学習タスク用に意図しているが、そのようなタスクのセットはもちろんやや広い。 実際、これらはSpark map-reduceには一般的ではないタスクです。 私が理解しているように、これらはほとんどが一度起動するか、クラッシュした場合のメッセージングコンポーネントです。



そのようなタスクをサポートするAPIが便利であれば、間違いなくその必要性があります。 私たちの会社では、そのようなコンポーネントはYarnアプリケーションとして設計されており、Sparkからは多少別々に動作します。 Spark内のより緊密で便利な統合は価値があります。

Avroサポートの改善

Avroのサポートは一般的に良好でした。 いくつかの追加のデータ型、つまり10進数、日付、時刻、期間などを含む、いわゆる「論理型」（実際にはいくつかの派生型）がサポートされています。



率直に言って、Hiveの作者（そして同時にSparkも）が寄木細工をよりよくサポートし、そのレイアウトに基づいてテーブルを作成する方法を学ぶとき、私はもっと待ちます。 これが可能になりましたが、Avroでは見た目も動作もより便利になりました。



ここで詳細を読むことができます 。

Scala 2.12のサポート（実験的）

Javaプログラマーとしては重要ではないように思えますが、このプロジェクトのフレームワーク内では、Java 8との相互作用、たとえばラムダのシリアル化を改善することを約束しました。