問題タブ [graphframes]

私はconnectedComponents()pyspark のグラフフレームから使用して、約 1800K の頂点と 500k のエッジを持つかなり大きなグラフの接続コンポーネントを計算しようとしていました。

6時間経っても課題は終わらない。Windowsを搭載した単一のマシンでpysparkを実行しています

を。与えられた設定でそのような計算を行うことは実行可能ですか?

b. 次のような警告メッセージが表示されました

これは何を意味するのでしょうか？

c. グラフフレームでグラフが無向であることをどのように指定できますか? 両方向にエッジを追加する必要がありますか?

次の問題があります:列Aの各IDごとに列Bの値のすべての組み合わせを見つけ、結果をDataFrameとして返す必要があります

以下の入力 DataFrame の例

次の出力 DataFrame を取得する必要があります (GraphX\GraphFrame 用です)。

私が今まで考えていた1つの解決策は次のとおりです。

出力: [(1, [(20,15),(30,20),(30,15)]),(5,[(10,14)]),(3,[(50,33)]) ]

そして、ここで私は立ち往生しています:(必要なデータフレームに戻す方法は? 1つのアイデアは、並列化を使用することでした:

spark_sc.pyspark_scという名前の他のファイルがあるため

しかし、私のコードは失敗しました:

spark_sc.sc()notを使用するmap()と、動作します。

最後のステップで何が欠けているのか分かりますか? 使用することはまったく可能parallelize()ですか？または、まったく別のソリューションが必要ですか？ありがとう！

現在、私たちの仕事では型付きデータセットを使用しています。現在、Graphframes を使用して調査しています。

ただし、Graphframes は Dataset[Row] である Dataframe に基づいているようです。Graphframes は型指定されたデータセットと互換性がありますか。例: データセット[人]

私は GraphFrames と Scala にかなり慣れていません。私はある種のラベル伝播アルゴリズムを書いています（ライブラリのものとは非常に異なります）。基本的に、各頂点には配列「memVector」があり、エッジには float 値「floatWeights」があります。各頂点の memVector を更新して、すべての隣接頂点からの (floatWeights * memVector) の合計にします。これは私が同じために書いたコードです:

配列と浮動小数点数を直接掛けることができないため、私が書いた aggfunc は正しくありません。上記をspark-shellで実行していますが、最後の行で次のエラーが発生しています:

私はそれに近づいていますか？回避策/解決策は大歓迎です。

Dataproc Spark クラスタでは、graphframe パッケージは spark-shell で使用できますが、jupyter pyspark ノートブックでは使用できません。

Pyspark カーネル構成:

以下は、クラスターを初期化するためのコマンドです。

次の SBT ファイルがあります。Apache GraphFrame を使用して Scala コードをコンパイルし、CSV ファイルも読み込んでいます。

これがScalaの私のコードです

SBT を使用して Jar ファイルを作成しようとすると、コンパイル中に次のエラーが発生します。

次のコードでは、Scala を使用して Apache Spark-GraphFrames を使用しています。上記のコードに BFS を適用し、頂点 0 から 100 までの距離を見つけようとしています。

ソース ノード:0 宛先ノード:100

頂点リストを以下に示します

エッジ一覧はこちら

しかし、上記のコードの問題点は、0 から 100 の頂点の実行だけでかなりの時間がかかることです。4 時間実行したのに出力がありません。上記のコードは、12 GB RAM を搭載した単一マシンで実行しています。

コードを高速化して最適化する方法を教えてください。