Launching and managing applications for Spark and PySpark

There are multiple ways to run a Spark or PySpark job in a Yandex Data Processing cluster:

• Spark Shell (a command shell for Scala and Python programming languages). This method runs calculations line by line, rather than using a script. For more information about Spark Shell, see the Spark documentation.
• spark-submit script. Saves the calculation results to HDFS. For more information about spark-submit, see the Spark documentation.
• Yandex Cloud CLI commands. These allow you to save calculation results not only to HDFS but also to a Yandex Object Storage bucket.

Below is an example demonstrating the calculation of 2018 US air traffic statistics based on data from transtats.bts.gov. The data set is stored in Parquet format in a public Yandex Object Storage bucket named yc-mdb-examples.

Getting startedGetting started

Prepare the infrastructure:

Using Spark ShellUsing Spark Shell

1. Use SSH to connect to the Yandex Data Processing cluster's master host.

2. Run Spark Shell on the master host:

   /usr/bin/pyspark
   

   The number of cores and executors is only limited by your Yandex Data Processing configuration.

3. Enter the following code line by line:

   sql = SQLContext(sc)
   df = sql.read.parquet("s3a://yc-mdb-examples/dataproc/example01/set01")
   

   The last line reads the data from the public bucket containing the sample data set. After this line is executed, an organized data set df (DataFrame) containing the data read becomes available for the current session.

4. To see the schema of the resulting DataFrame, run the command:

   df.printSchema()
   

   The terminal displays a list of columns with their types.

5. Calculate flight statistics by month and find the top ten cities by the number of departures:

   • Number of flights by month:

     df.groupBy("Month").count().orderBy("Month").show()
     

   • Top ten cities by number of departures:

     df.groupBy("OriginCityName").count().orderBy("count", ascending=False).show(10)
     

Using Spark SubmitUsing Spark Submit

Spark Submit allows you to run pre-written applications using the spark-submit script. In this example, we will calculate the number of flights by month.

Terminating the applicationTerminating the application

By default, the resources of the running application are managed by the YARN component. If you need to terminate or remove the application from the queue, use the yarn utility:

1. List the applications:

   yarn application -list
   

2. Terminate the application you no longer need:

   yarn application -kill <app_ID>
   

For more information about YARN commands, see YARN Commands.

Running jobs using the Yandex Cloud CLIRunning jobs using the Yandex Cloud CLI

Jobs are run using the Yandex Cloud CLI through the Yandex Data Processing agent installed on the cluster master host. The agent gets job parameters through the Yandex Data Processing API.

The executable file and its dependencies must be located in a storage accessible to the Yandex Data Processing cluster service account. The executed application itself must have access to the storages in which the source data set and execution results are saved.

You can save the calculation results to HDFS on the Yandex Data Processing cluster or to the data-proc-bucket bucket you specified when creating the cluster.

All service and debugging information will be saved to data-proc-bucket. For each job, the Yandex Data Processing agent creates a separate folder with a path, such as dataproc/clusters/<cluster_ID>/jobs/<job_ID>.

Below are the two application versions, one for Python and one for Scala.

Running a PySpark jobRunning a PySpark job

To run a PySpark job:

1. Install additional dependencies.
2. Prepare and run a PySpark job.

Install additional dependenciesInstall additional dependencies

On a local computer:

1. If you do not have the Yandex Cloud command line interface yet, install and initialize it.

   The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name or --folder-id parameter.

2. Install and configure the S3cmd console client to work with Yandex Object Storage.

3. Install Python. Make sure the Python version matches the version available from the image. You can check the version under Runtime environment. For image version 2.0, use Python 3.8.10:

   sudo apt update && sudo apt install python3.8
   

Prepare and run a PySpark jobPrepare and run a PySpark job

1. Create a file named job.py with the following code:

   import sys
   
   from pyspark import SparkContext, SparkConf
   from pyspark.sql import SQLContext
   
   def main():
   
       if len(sys.argv) != 3:
           print('Usage job.py <input_directory> <output_directory>')
           sys.exit(1)
   
       in_dir = sys.argv[1]
       out_dir = sys.argv[2]
   
       conf = SparkConf().setAppName('Month Stat - Python')
       sc = SparkContext(conf=conf)
       sql = SQLContext(sc)
       df = sql.read.parquet(in_dir)
       month_stat = df.groupBy('Month').count()
       job_id = dict(sc._conf.getAll())['spark.yarn.tags'].replace('dataproc_job_', '')
       if out_dir.startswith('s3a://'):
           month_stat.repartition(1).write.format('csv').save(out_dir + job_id)
       else:
           default_fs = sc._jsc.hadoopConfiguration().get('fs.defaultFS')
           month_stat.repartition(1).write.format('csv').save(default_fs + out_dir + job_id)
   
   
   if __name__ == '__main__':
       main()
   

2. To make sure that PySpark has access to your code, upload the job.py file to the Object Storage bucket that the Yandex Data Processing cluster service account has access to:

   s3cmd put ./job.py s3://data-proc-bucket/bin/
   

3. Run the job.

   The run command varies depending on whether you want to save the job results to Object Storage or to HDFS.

   Object Storage

   HDFS directory

   yc dataproc job create-pyspark \
      --cluster-id=<cluster_ID> \
      --name=<job_name> \
      --main-python-file-uri="s3a://data-proc-bucket/bin/job.py" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="s3a://data-proc-bucket/jobs_results/"
   

   In the command, specify the following:

   • --cluster-id: Cluster ID. You can get it with a list of clusters in the folder.
   • --name: Arbitrary Spark job name.

   A CSV file with the execution results is saved to data-proc-bucket.

   yc dataproc job create-pyspark \
      --cluster-id=<cluster_ID> \
      --name=<job_name> \
      --main-python-file-uri="s3a://data-proc-bucket/bin/job.py" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="tmp/jobs/"
   

   In the command, specify the following:

   • --cluster-id: Cluster ID. You can get it with a list of clusters in the folder.
   • --name: Arbitrary Spark job name.

   A CSV file with the execution result is saved to the /tmp/jobs/<job_ID>/ folder in HDFS.

4. (Optional) View the job logs:

   yc dataproc job log <job_ID> --cluster-id=<cluster_ID>
   

Running Spark jobsRunning Spark jobs

To run a Spark job:

1. Install additional dependencies.
2. Build a Scala application.
3. Upload the JAR file to Object Storage.
4. Run the Spark job in the Yandex Data Processing cluster.

Install additional dependenciesInstall additional dependencies

1. If you do not have the Yandex Cloud command line interface yet, install and initialize it.

   The folder specified in the CLI profile is used by default. You can specify a different folder using the --folder-name or --folder-id parameter.

2. Use SSH to connect to the Yandex Data Processing cluster's master host.

3. Install sbt, the standard build utility for Scala. It comes bundled with the Scala programming language.

4. Install and configure the S3cmd console client to work with Yandex Object Storage.

Build a Scala applicationBuild a Scala application

To simplify dependency management, build the application to a single JAR file (fat JAR) using the sbt-assembly plugin:

1. Create a folder named spark-app with the project and src/main/scala subdirectories.

2. Create a file named spark-app/project/plugins.sbt that describes the connection of the sbt-assembly plugin to assemble a single JAR file:

   addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "<sbt_assembly_version>")
   

   You can check the version of the sbt-assembly plugin in its repository, under Releases.

3. Run the scala -version command to learn the version of Scala installed on your system.

4. Create a file named spark-app/build.sbt with a description of the dependencies and the strategy for merging them into a single JAR file. Specify the Scala version in the build.sbt file:

   scalaVersion := "<Scala_version>"
   
   libraryDependencies ++= Seq(
       "org.apache.spark" %% "spark-core" % "2.4.4",
       "org.apache.spark" %% "spark-sql" % "2.4.4",
   )
   
   assembly / assemblyMergeStrategy := {
     case PathList("org","aopalliance", xs @ _*) => MergeStrategy.last
     case PathList("javax", "inject", xs @ _*) => MergeStrategy.last
     case PathList("javax", "servlet", xs @ _*) => MergeStrategy.last
     case PathList("javax", "activation", xs @ _*) => MergeStrategy.last
     case PathList("org", "apache", xs @ _*) => MergeStrategy.last
     case PathList("com", "google", xs @ _*) => MergeStrategy.last
     case PathList("com", "esotericsoftware", xs @ _*) => MergeStrategy.last
     case PathList("com", "codahale", xs @ _*) => MergeStrategy.last
     case PathList("com", "yammer", xs @ _*) => MergeStrategy.last
     case "about.html" => MergeStrategy.rename
     case "overview.html" => MergeStrategy.last
     case "META-INF/ECLIPSEF.RSA" => MergeStrategy.last
     case "META-INF/mailcap" => MergeStrategy.last
     case "META-INF/mimetypes.default" => MergeStrategy.last
     case "plugin.properties" => MergeStrategy.last
     case "log4j.properties" => MergeStrategy.last
     case "git.properties" => MergeStrategy.last
     case x =>
       val oldStrategy = (assembly / assemblyMergeStrategy).value
       oldStrategy(x)
   }
   

5. Create a file named spark-app/src/main/scala/app.scala with the application code:

   package com.yandex.cloud.dataproc.scala
   
   import org.apache.spark.{SparkConf, SparkContext}
   import org.apache.spark.sql.SQLContext
   
     object Main {
       def main(args: Array[String]) {
         if (args.length != 2){ //check the argument
           System.err.println("Usage spark-app.jar <input_directory> <output_directory>");
           System.exit(-1);
         }
         val inDir = args(0); //URI to source data
         val outDir = args(1); //URI to the output directory
         val conf = new SparkConf().setAppName("Month Stat - Scala App")
         val sc = new SparkContext(conf)
         val sqlContext = new org.apache.spark.sql.SQLContext(sc)
         val df = sqlContext.read.parquet(inDir)
         val monthStat = df.groupBy("Month").count()
         val defaultFS = sc.hadoopConfiguration.get("fs.defaultFS") //get the HDFS server endpoint
         val jobId = conf.get("spark.yarn.tags").replace("dataproc_job_", ""); //get the job ID
         if (outDir.toLowerCase().startsWith("s3a://")) {
           monthStat.repartition(1).write.format("csv").save(outDir + jobId)
         } else {
           monthStat.repartition(1).write.format("csv").save(defaultFS + "/" + outDir + jobId)
         }
   
         sc.stop()
       }
     }
   

6. Run the app build in the spark-app folder:

   sbt clean && sbt compile && sbt assembly
   

   If you get the Error looking up function 'stat' error

   If you get the java.lang.UnsatisfiedLinkError: Error looking up function 'stat': java: undefined symbol: stat error and your master host OS is Ubuntu, run each sbt command with the -Dsbt.io.jdktimestamps=true flag:

   sbt clean -Dsbt.io.jdktimestamps=true && \
   sbt compile -Dsbt.io.jdktimestamps=true && \
   sbt assembly -Dsbt.io.jdktimestamps=true
   

The file will be available at the following path: spark-app/target/scala-<Scala_version>/spark-app-assembly-0.1.0-SNAPSHOT.jar.

Upload the JAR file to Object StorageUpload the JAR file to Object Storage

For Spark to have access to the assembled JAR file, upload the file to data-proc-bucket. You can upload the file using s3cmd:

s3cmd put ~/spark-app/target/scala-<Scala_version>/spark-app-assembly-0.1.0-SNAPSHOT.jar s3://data-proc-bucket/bin/

The file is uploaded to s3://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar.

Run the Spark job in the Yandex Data Processing clusterRun the Spark job in the Yandex Data Processing cluster

1. Disconnect from the cluster master host.

2. Run the job.

   The run command varies depending on whether you want to save the job results to Object Storage or to HDFS.

   Object Storage

   HDFS directory

   yc dataproc job create-spark \
      --cluster-id=<cluster_ID> \
      --name=<job_name> \
      --main-class="com.yandex.cloud.dataproc.scala.Main" \
      --main-jar-file-uri="s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="s3a://data-proc-bucket/jobs_results/"
   

   In the command, specify the following:

   • --cluster-id: Cluster ID. You can get it with a list of clusters in the folder.
   • --name: Arbitrary Spark job name.

   A CSV file with the execution results is saved to data-proc-bucket.

   yc dataproc job create-spark \
      --cluster-id=<cluster_ID> \
      --name=<job_name> \
      --main-class="com.yandex.cloud.dataproc.scala.Main" \
      --main-jar-file-uri="s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar" \
      --args="s3a://yc-mdb-examples/dataproc/example01/set01" \
      --args="tmp/jobs/"
   

   In the command, specify the following:

   • --cluster-id: Cluster ID. You can get it with a list of clusters in the folder.
   • --name: Arbitrary Spark job name.

   A CSV file with the execution result is saved to the /tmp/jobs/<job_ID>/ folder in HDFS.

   Example message saying that the job was run successfully:

   done (1m2s)
   id: {your_job_id}
   cluster_id: {your_cluster_id}
   name: test02
   status: DONE
   spark_job:
     args:
     - s3a://yc-mdb-examples/dataproc/example01/set01
     - s3a://data-proc-bucket/jobs_results/
     main_jar_file_uri: s3a://data-proc-bucket/bin/spark-app-assembly-0.1.0-SNAPSHOT.jar
     main_class: com.yandex.cloud.dataproc.scala.Main
   

Delete the resources you createdDelete the resources you created

Some resources are not free of charge. To avoid paying for them, delete the resources you no longer need: