スパークの「ジョブ」を単一のアプリケーションに送信する方法を考えています（RDD作業の共有だけでなく、モジュールのコード/実行の独立性も可能にします）。ooyala でこのspark-jobserverを正式に見たことがありますが、まだ Python をサポートしていないことに気付きました。これは、scala/java でのこのユースケースで取られる一般的なルートですか? それとも、ここで間違ったルートを下っていますか？

また、人気の低いpyspark-jobserverと、python と R に対処するためのメインの spark-jobserver に関する未解決の問題も確認しました。

spark-jobserver の使用例をよりよく理解するために、詳細なジョブ スケジューリング フレームワークを考えると、なぜこの機能が spark によって直接サポートされないのか疑問に思っています。

スパークを使い始める方法flatmapまたはデータフレームを知りたいexplodeです。

を使用して作成されdf.groupBy("columName").count、収集すると次の構造になります。

しかし、私はむしろ次のようなものが欲しいです

このようなことを達成するための適切なツールは何ですか? フラットマップ、エクスプロージョンか何か？

コンテキスト: spark-jobserver を使用したいです。後者の形式でデータを提供した場合にのみ、意味のある結果（たとえば、機能するjsonシリアル化）を提供するようです

spark-jobserver を使い始めて、Sparkがデータフレームをフラット化するようにデータ フレームをフラット化できることを学びましたが、これはまだhttps://github.com/spark-jobserver/spark-jobserver#job-result-serializationを満たしません

これがスパークから得た結果である場合

どうすればそれを適切な形式にマッピングできますか? (便利なシリアル化) フィールドを追加するにはどうすればよいですか?

次のような配列で遊んでみようとするとArray([1,2], [3,4])、エラーが発生するだけです。

現在、データフレームをフラット化する Sparkに基づいて、次のシリアル化を取得します。

これは明らかにジョブサーバーによって「解析」されていません。

私が理解している限り、ネストされた配列 (からcollect) は正しくシリアル化できません。ただし、このマップはシリアライズ可能である必要があります。なにが問題ですか？

編集

適切に型指定されたリストを返す場合にのみ、Json エンコーディングが機能するようです。

結果は次のとおり"result": [["Torcuato", 27], ["Rosalinda", 34]]です。すでに

"result": ["[Torcuato,27]", "[Rosalinda,34]"]奇妙な種類の Json を取得します。

問題を理解している限り、すべての結果をカスタム クラスに解析する必要があります。どうすればこれを達成できますか？

Kafkaからデータを読み取り、変換後にこのデータをhttpエンドポイント（または別のkafka - この質問ではhttpを考えてみましょう）に送信する単純なスパークストリーミングアプリケーションがあります。job-serverを使用してジョブを送信しています。

現在、「auto.offset.reset」="smallest" および interval=3s でソース kafka から消費を開始しています。幸せな場合、すべてが良さそうに見えます。ここに抜粋があります：

"auto.offset.reset"="smallest" であるため、これは 1 つのジョブで約 200K のメッセージを処理します。ジョブの途中で http サーバーを停止し (POST で問題をシミュレート)、httpProcessor.process が例外をスローすると、そのジョブは失敗し、未処理のものはすべて失われます。その後、3秒ごとにポーリングし続けていることがわかります。

だから私の質問は：

1. 次の 3 秒のジョブで X メッセージを受け取り、エラーが発生する前に Y のみを処理できた場合、残りの XY は処理されないという私の仮定は正しいですか?
2. Kafka からのストリーム/消費を一時停止する方法はありますか? たとえば、断続的なネットワークの問題が発生し、消費されたすべてのメッセージがその間に失われる可能性が高い場合です。再試行を続けるもの (おそらく指数バックオフ) で、http エンドポイントがアップするたびに、再び消費を開始します。

ありがとう

シングルモードで Spark クラスターを起動しました。

Scala 用の spark-core lib を使用してジョブを正常に実行できます。ジョブ管理に Spark JobServer を使用したいと考えています。マスターホストノードのdockerで開始しました：

次に、既に追加されたジョブを実行してみます (追加されたことを UI で確認しました)。

応答：

Job Server にログオンします。

Spark バージョン: 1.6.1 Hadoop 2.6 用に
事前構築済み Spark JobServer: 0.6.0
Job Scala バージョン 2.10.6

Cassandra に接続する Spark ジョブ サーバーを作成しています。レコードを取得した後、単純なグループ化を実行して合計したいと思います。データを取得できましたが、出力を印刷できませんでした。私は何時間もグーグルを試し、cassandra グーグルグループにも投稿しました。私の現在のコードは以下の通りで、収集時にエラーが発生しています。

}

私は Spark の完全な初心者であり、これについてさらに詳しく調べ始めたところです。CDHディストリビューションを使用してhadoopをインストールしないことで、より長いパスを選択し、Apache WebサイトからHadoopをインストールし、構成ファイルを自分で設定して、基本をさらに理解しました。

3 ノード クラスターをセットアップしました (すべてのノードは ESX サーバーから作成された VM マシンです)。Zookeeper メカニズムを使用して、Namenode と ResourceManager の両方に高可用性をセットアップしました。3 つのノードはすべて DataNode としても使用されています。

次のデーモンが 3 つのノードすべてで実行されています

NameNode 1 と 2 で NN と RM の HA をセットアップしました。ノードは非常に最小限のハードウェア構成 (それぞれ 4GM の RAM と 20GB のディスク容量) ですが、これらはテスト目的のためだけのものです。だから大丈夫だと思います。

NameNode 1 に Spark (インストール済みの Hadoop 2.7 と互換性のあるバージョン) をインストールしました。Spark-shell をローカルで起動し、基本的な scala コマンドを実行して RDD を作成し、それに対していくつかのアクションを実行できます。また、SparkPi の例を Yarn-Cluster および Yarn-Client 展開モードとしてテスト実行することもできます。すべてがうまく機能します。

今私の問題は、リアルタイムのシナリオで、(Java、scala、または py) ベースのコードをローカル マシンに記述します (Hadoop クラスターを形成するノードではありません)。HA クラスターと同じネットワークに別のマシンがあるとします。HA にないホストから Yarn RM にジョブを送信するにはどうすればよいですか (SparkPi を送信してみます)。私はこれをします ？

コードを書いているマシンに SPARK をインストールする必要があり (私の仮定は正しいですか)、HA クラスターに Spark をインストールする必要はありません。また、送信されたジョブの出力を、送信された場所からホストに戻したいと考えています。これを機能させるために何をする必要があるかわかりません。

Spark JobServerについて聞いたことがありますが、これはすべてを起動して実行するために必要なものですか? 皆さんがこの混乱で私を助けてくれると信じています。これを行うための手順を明確に指定しているドキュメントが見つかりませんでした。Windows ベースのマシンから UNIX 環境の HA クラスター セットアップにジョブを送信できますか?