SparkでPythonApacheBeamパイプラインを実行する

Question

ここでApacheビーム（python sdkを使用）を試してみたので、単純なパイプラインを作成して、Sparkクラスターにデプロイしようとしました。

from apache_beam.options.pipeline_options import PipelineOptions
import apache_beam as beam

op = PipelineOptions([
        "--runner=DirectRunner"
    ]
)


with beam.Pipeline(options=op) as p:
    p | beam.Create([1, 2, 3]) | beam.Map(lambda x: x+1) | beam.Map(print)

このパイプラインはDirectRunnerでうまく機能しています。したがって、同じコードをSparkにデプロイするには（移植性はBeamの重要な概念であるため）...

最初に私PipelineOptionsはここで述べたように編集しました：

op = PipelineOptions([
        "--runner=PortableRunner",
        "--job_endpoint=localhost:8099",
        "--environment_type=LOOPBACK"
    ]
)

job_endpoint次のコマンドを使用して実行するビームスパークジョブサーバーのDockerコンテナへのURLです。

docker run --net=host apache/beam_spark_job_server:latest --spark-master-url=spark://SPARK_URL:SPARK_PORT

これはうまく機能するはずですが、Sparkではこのエラーでジョブが失敗します：

20/10/31 14:35:58 ERROR TransportRequestHandler: Error while invoking RpcHandler#receive() for one-way message.

java.io.InvalidClassException: org.apache.spark.deploy.ApplicationDescription; local class incompatible: stream classdesc serialVersionUID = 6543101073799644159, local class serialVersionUID = 1574364215946805297

また、beam_spark_job_serverログに次の警告があります。

WARN org.apache.beam.runners.spark.translation.SparkContextFactory: Creating a new Spark Context.

ここで問題がどこにあるのか考えはありますか？コンテナ化されたサービスを通過せずにSparkでPythonビームパイプラインを実行する他の方法はありますか？

Answer 1

これは、ジョブサーバーに含まれているSparkクライアントのバージョンとジョブの送信先のSparkのバージョンが一致しないために発生する可能性があります。