Apache sparkのクラスタノード間で独立して処理される異なるグラフファイルを処理するにはどうすればいいですか？

Question

多くのグラフファイルがあり、各グラフには約500Kのエッジがあります。私はApache Sparkでこれらのグラフファイルを処理していましたが、グラフ処理ジョブ全体を効率的に並列化する方法は不思議でした。今のところ、すべてのグラフファイルは他のファイルと独立しているので、ファイルとの並列性を探しています。したがって、グラフファイルが100個あり、ノードクラスタが20個ある場合、各ノードで各ファイルを処理できるので、各ノードは5つのファイルを処理します。さて、何が起こっているかは、単一のグラフが多数の段階で処理されていて、シャッフルが多く発生しているようなものです。

graphFile = "/mnt/bucket/edges" #This directory has 100 graph files each file with around 500K edges 

nodeFile = "/mnt/bucket/nodes" #This directory has node files 

graphData = sc.textFile(graphFile).map(lambda line: line.split(" ")).flatMap(lambda edge: [(int(edge[0]),int(edge[1]))]) 

graphDataFrame = sqlContext.createDataFrame(graphData, ['src', 'dst']).withColumn("relationship", lit('edges')) # Dataframe created so as to work with Graphframes 

nodeData = sc.textFile(nodeFile).map(lambda line: line.split("\s")).flatMap(lambda edge: [(int(edge[0]),)]) 

nodeDataFrame = sqlContext.createDataFrame(nodeData, ['id']) 

graphGraphFrame = GraphFrame(nodeDataFrame, graphDataFrame) 

connectedComponent = graphGraphFrame.connectedComponents() 

2つのファイルでも処理するには時間がかかります。そして私は20Kファイルのように処理しなければなりません。各ファイルには800Kのエッジがあります。すべての従属エッジが単一ノードで処理されることを確実にするデータパーティション戦略を把握できれば、シャッフルは少なくなります。

これを効率的に解決するにはどうすればよいですか？

Answer 1

TL; DR Apache Sparkは適切なツールではありません。

スパークの主な範囲はデータの並列処理ですが、あなたが探しているのはタスクの並列処理です。理論的にコアのSparkエンジンは、限られたタスクの並列性を達成するのに十分な汎用性がありますが、実際にはこのような仕事のための優れたツールがあり、GraphXやGraphFramesのようなライブラリの目標は間違いありません。

これらのライブラリの背後にある主要な前提はデータの配布であるため、アルゴリズムはメッセージの受け渡しや多段階のジョブ構造とシャッフルに反映されるような手法を使用して実装されます。データがメインメモリに収まる場合（最適化されたグラフ処理ライブラリを使用して単一ノード上の何百万ものエッジを持つグラフを簡単に処理できます）、これらの手法は実際にはまったく役に立たないものです。あなたが示されてきたコードの一部を考える

は

、炉心内igraphまたはNetworkXのようなグラフ処理ライブラリ十二分とはるかにする必要があります（より良い文書化され、はるかに包括的な残念ながらメモリ空腹とややslowish）GNU Parallelと組み合わせます実際には効率的です。複雑なジョブの場合は、AirflowやLuigiのようなフル機能のワークフロー管理ツールの使用を検討することもできます。