問題タブ [apache-spark-2.0]

Spark ストリーミングを使用して kafka からメッセージを読み取る Spark 2.0 アプリケーションがあります (spark-streaming-kafka-0-10_2.11 を使用)。

構造化されたストリーミングはとてもクールに見えるので、コードを移行してみようと思ったのですが、使い方がわかりません。

通常のストリーミングでは kafkaUtils を使用して createDstrean を使用し、渡したパラメーターでは値デシリアライザーでした。

構造化ストリーミングでは、ドキュメントには DataFrame 関数を使用して逆シリアル化する必要があると書かれていますが、それが何を意味するのか正確にはわかりません。

この例などの例を見ましたが、Kafka の Avro オブジェクトは非常に複雑で、例の String のように単純にキャストすることはできません..

これまでのところ、この種のコードを試しました（別の質問でここで見ました）：

そして、「データ型の不一致: BinaryType を StructType(StructField(....) にキャストできません」

値を逆シリアル化するにはどうすればよいですか?

Dataset<Row>json 文字列の単一の列を持つ:

Json サンプル:

次のように最も効率的に取得するにはどうすればよいですかDataset<Row>。

これらのデータをストリームで処理しています。ファイルから読み取っているときに、spark が自分でこれを実行できることを知っています。

しかし今、私はカフカからデータを読んでおり、別の形式でデータを提供しています。私は Gson のようないくつかのパーサーを使用できることを知っていますが、spark にそれをさせたいと思っています。

I'm running Spark 2.0 on Standalone mode, successfully configured it to launch on a server and also was able to configure Ipython Kernel PySpark as option into Jupyter Notebook. Everything works fine but I'm facing the problem that for each Notebook that I launch, all of my 4 workers are assigned to that application. So if another person from my team try to launch another Notebook with PySpark kernel, it simply does not work until I stop the first notebook and release all the workers.

To solve this problem I'm trying to follow the instructions from Spark 2.0 Documentation. So, on my $SPARK_HOME/conf/spark-defaults.conf I have the following lines:

Also, on $SPARK_HOME/conf/spark-env.sh I have:

But when I try to launch the workers, using $SPARK_HOME/sbin/start-slaves.sh, only the first worker is successfully launched. The log from the first worker end up like this:

16/11/24 13:32:06 INFO Worker: Successfully registered with master spark://cerberus:7077

But the log from workers 2-4 show me this error:

INFO ExternalShuffleService: Starting shuffle service on port 7337 with useSasl = false 16/11/24 13:32:08 ERROR Inbox: Ignoring error java.net.BindException: Address already in use

It seems (to me) that the first worker successfully starts the shuffle-service at port 7337, but the workers 2-4 "does not know" about this and try to launch another shuffle-service on the same port.

The problem occurs also for all workers (1-4) if I first launch a shuffle-service (using $SPARK_HOME/sbin/start-shuffle-service.sh) and then try to launch all the workers ($SPARK_HOME/sbin/start-slaves.sh).

Is any option to get around this? To be able to all workers verfy if there is a shuffle service running and connect to it instead of try to create a new service?

スタンドアロン モードで正常に実行されていた Spark アプリケーションがあり、同じアプリケーションを AWS EMR クラスターで実行しようとしていますが、現在失敗しています。

このメッセージは、これまでに見たことのないもので、ワーカーが仕事を受け取っておらず、シャットダウンされていることを意味しています。

DAG は、初期化されたワーカー、次に収集 (比較的小さいもの)、そしてすべてが失敗した直後を示しています。動的割り当てが有効になっているため、おそらくドライバーがタスクを送信していないためにタイムアウトになったのではないかと考えられました-理論を証明するために、動的割り当てなしで別のクラスターをスピンアップしましたが、同じことが起こりました.

マスターは糸に設定されています。

どんな助けでも大歓迎です、ありがとう。

私のステップは非常に簡単です -spark-submit --deploy-mode client --master yarn --class Run app.jar

私はこのようなデータフレームを変換しています

次に、3 つの列で構成され、次のようなデータフレームに変換する必要があります。

randomSplit メソッドを使用してランダムに分割できることはわかっています。

「nonRandomSplit メソッド」を使用して、データを連続した部分に分割できますか?

アパッチ スパーク 2.0.1。前もって感謝します。

UPD: データの順序は重要です。「ID が小さい」データでモデルをトレーニングし、ID が大きいデータでモデルをテストします。そのため、データをシャッフルせずに連続した部分に分割したいと考えています。

例えば

私が考えることができる唯一の解決策は、countとlimitを使用することですが、おそらくもっと良い方法があります。

Spark 2.0.x では、JobProgressListener実装を使用して、クラスターからジョブ/ステージ/タスクの進行状況情報をリアルタイムで取得しています。イベント フローがどのように機能するかを理解しており、作業に関する最新情報を正常に受け取ることができます。

私の問題は、同じ Spark コンテキストで同時に複数の異なる送信を実行していることです。各送信に属するジョブ/ステージ/タスクを区別することは一見不可能です。各ジョブ/ステージ/タスクは一意の ID を受け取ります。これは素晴らしいことです。ただし、受信した JobProgressListener イベント オブジェクトと一緒に返される送信 "id" または "name" を提供する方法を探しています。

「ジョブ グループ」は Spark コンテキストで設定できることは認識していますが、複数のジョブが同じコンテキストで同時に実行されていると、それらがスクランブルされてしまいます。

単一の SQLContext のリスナー イベントで返されるカスタム プロパティを忍び込ませる方法はありますか? そうすることで、後続の Stage および Task イベントをリンクして、必要なものを取得できるはずです。

注意: 私はこれらのジョブに spark-submit を使用していません。これらは、SparkSession/SQLContext への Java 参照を使用して実行されています。

解決策やアイデアをありがとう。