Exécution d'un pipeline Apache Beam Python sur Spark

Question

Je suis en train d'essayer Apache beam (avec python sdk) ici, j'ai donc créé un pipeline simple et j'ai essayé de le déployer sur un cluster 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)

Ce pipeline fonctionne bien avec DirectRunner. Donc pour déployer le même code sur Spark (car la portabilité est un concept clé dans Beam) ...

J'ai d'abord édité le PipelineOptionscomme mentionné ici :

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

job_endpointest l'URL du conteneur docker du serveur de travaux beam spark que j'exécute à l'aide de la commande:

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

Cela est censé fonctionner correctement mais le travail échoue sur Spark avec cette erreur:

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

En outre, j'ai cet avertissement dans les beam_spark_job_serverjournaux:

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

Une idée où est le problème ici? Existe-t-il un autre moyen d'exécuter des pipelines python Beam sur Spark sans passer par un service conteneurisé?

Answer 1

Cela peut se produire en raison d'une incompatibilité de version entre la version du client Spark contenue dans le serveur de travaux et la version de Spark à laquelle vous soumettez le travail.