Differences

This shows you the differences between two versions of the page.

Link to this comparison view

Both sides previous revision Previous revision
Next revision 		Previous revision
doku:bigdata [2021/05/27 09:19]
dieter 		doku:bigdata [2023/05/11 08:52] (current)
katrin
Line 1: Line 1:
 ====== Big Data ====== ====== Big Data ======
  
-VSC has Big Data modules for running  +VSC-4 has Big Data modules for running  
-  * MapReduce jobs, and+  * MapReduce jobs 
 +    * in several programming languages, including Java, and  
 +    * using executables which can stream data from standard input to standard output (e.g. grep, cut, tr, sed, ...), and
   * Spark jobs using    * Spark jobs using 
     * Python,     * Python,
Line 10: Line 12:
     * SQL     * SQL
 either on standard file systems (using Spectrum Scale, formerly GPFS) or HDFS (Hadoop distributed file system) locally on the nodes of a job. either on standard file systems (using Spectrum Scale, formerly GPFS) or HDFS (Hadoop distributed file system) locally on the nodes of a job.
 +
 +==== Big Data and HPC ====
  
 Among the advantages of Big Data also on HPC environments like VSC, we want to mention Among the advantages of Big Data also on HPC environments like VSC, we want to mention
-  * very easy coding, also in data centric languages like R or SQL, +  * very easy coding, in many languages, including data centric ones like R or SQL, 
-  * automatic parallelization without any changes in the user program,+  * automatic parallelization without any changes in the user program, and
   * extremely good scaling behaviour.   * extremely good scaling behaviour.
  
 Users have to be aware of difficulties for running Big Data jobs on HPC clusters, namely of shared resources and granularity. Users have to be aware of difficulties for running Big Data jobs on HPC clusters, namely of shared resources and granularity.
-  * File systems are shared by all users. Users with high file system demand should use the local HDFS. +  * File systems are shared by all users. Users with high file system demand should use the local HDFS, if the same data is read more than once in a job. Please, do not submit many Big Data jobs at the same time, e.g. slurm array jobs! 
-  * Login nodes and internet connections are shared by all users. Bringing huge amounts of data to VSC takes some time. Please make sure that other users of VSC are not affected! +  * Login nodes and internet connections are shared by all users. Bringing huge amounts of data to VSC takes some time. Please make sure that other users of VSC are not affected! (One or two ssh-connections from your workstation are OK, whereas 10 simultaneous connections from a computer with high internet bandwidth, i.e. at least 10Mb/s, could affect others.) 
-  * Only very coarse grained parallelization will work with Big Data frameworks. Typically parallelization is done by partitioning input data and doing the majority of the calculation independently by separated processes. +  * Only very coarse grained parallelization will work with Big Data frameworks. Typically parallelization is done by partitioning input data (automatically) and doing the majority of the calculation independently by separated processes. 
  
-In the context of high performance computing a typical application of Big Data methods is pre- and/or postprocessing of large amounts of data, e.g. filtering.+In the context of high performance computing a typical application of Big Data methods is pre- and/or postprocessing of large amounts of data, e.g.  
 +  * filtering, 
 +  * data cleaning, 
 +  * discarding input fields which are not required, 
 +  * counting, 
 +  * ...
  
-===== Apache Spark on VSC =====+Frameworks and applications using Big Data methods are shared among many scientific communities.
  
-There are many frameworks in Data Science that are using <html><span style="color:#cc3300;font-size:100%;">&dzigrarr;</span> </html> [[https://spark.apache.org|Apache Spark]]<html><sup>TM</sup></html> , for instance, [[http://bdgenomics.org|ADAM]] for genomic analysis, [[https://geotrellis.io|GeoTrellis]] for geospatial data analysis, or [[https://koalas.readthedocs.io/en/latest/#|Koalas]] (distributed Pandas).+==== Apache Spark on VSC ====
  
-==== Example SLURM & Python scripts ====+There are many frameworks in Data Science that are using <html><span style="color:#cc3300;font-size:100%;">&dzigrarr;</span> </html> [[https://spark.apache.org|Apache Spark]]<html><sup>TM</sup></html> , for instance, [[https://adam.readthedocs.io|ADAM]] for genomic analysis, [[https://geotrellis.io|GeoTrellis]] for geospatial data analysis, or [[https://koalas.readthedocs.io/en/latest/#|Koalas]] (distributed Pandas).
  
-Let us calculate the value of pi by means of the following Python script <html><span style="color:#cc3300;font-size:100%;">&dzigrarr;</span> </html> ''pi.py'' (from the Spark distribution): +=== Example slurm & python scripts ===
-<code> +
-from __future__ import print_function+
  
 +Let us calculate the value of pi by means of the following Python script ''pi.py'' (example originally taken from the Spark distribution):
 +
 +<code python>
 +from operator import add
 +from random import random
 import sys import sys
-from random import random + 
-from operator import add +
 from pyspark.sql import SparkSession from pyspark.sql import SparkSession
 + 
 if __name__ == "__main__": if __name__ == "__main__":
     """     """
Line 47: Line 57:
         .appName("PythonPi")\         .appName("PythonPi")\
         .getOrCreate()         .getOrCreate()
 + 
     partitions = int(sys.argv[1]) if len(sys.argv) > 1 else 2     partitions = int(sys.argv[1]) if len(sys.argv) > 1 else 2
     n = 100000 * partitions     n = 100000 * partitions
 + 
     def f(_):     def f(_):
         x = random() * 2 - 1         x = random() * 2 - 1
         y = random() * 2 - 1         y = random() * 2 - 1
         return 1 if x ** 2 + y ** 2 <= 1 else 0         return 1 if x ** 2 + y ** 2 <= 1 else 0
- +  
-    count = spark.sparkContext.parallelize(range(1, n + 1), partitions).map(f).reduce(add) +    count = spark.sparkContext\ 
-    print("Pi is roughly %f" % (4.0 * count / n)) +        .parallelize(range(1, n + 1), partitions)\ 
- +        .map(f)\ 
-    spark.stop()</code>+        .reduce(add) 
 +    pi = 4.0 * count / n 
 +    print(f"Pi is roughly {pi}") 
 +  
 +    spark.stop() 
 +</code>
 The idea of the calculation is to check (very often) whether a random point in the area [-1,-1; 1,1] is within a unit circle (i.e. has a distance less than 1 from origin). Since we know formulas for the area of the square as well as the circle we can estimate pi. The idea of the calculation is to check (very often) whether a random point in the area [-1,-1; 1,1] is within a unit circle (i.e. has a distance less than 1 from origin). Since we know formulas for the area of the square as well as the circle we can estimate pi.
  
-In order to submit a batch job to the queuing system we use the SLURM script <html><span style="color:#cc3300;font-size:100%;">&dzigrarr;</span> </html> ''pi.slrm'':+Assuming that the script is saved to ''$HOME/pi.py''' we can use the following SLURM script ''pi.slrm'' to run the code on VSC-4:
  
-<code>+<code bash>
 #!/bin/bash #!/bin/bash
 #SBATCH --nodes=1 #SBATCH --nodes=1
Line 71: Line 86:
 #SBATCH --error=err_spark-yarn-pi-%A #SBATCH --error=err_spark-yarn-pi-%A
 #SBATCH --output=out_spark-yarn-pi-%A #SBATCH --output=out_spark-yarn-pi-%A
 +#SBATCH --partition=skylake_0096
 +#SBATCH --qos=skylake_0096
  
 module purge module purge
-module load python/3.8.0-gcc-9.1.0-wkjbtaa + 
-module load openjdk/11.0.2-gcc-9.1.0-ayy5f5t+# you can choose one of the following python versions 
 +# with the current hadoop/spark installation on VSC-4 
 +# note: the current spark version does NOT work with python 3.11 
 +module load python/3.10.7-gcc-12.2.0-5a2kkeu 
 +# module load python/3.9.13-gcc-12.2.0-ctxezzj 
 +# module load python/3.8.12-gcc-12.2.0-tr7w5qy 
 + 
 +module load openjdk
 module load hadoop module load hadoop
 module load spark module load spark
 +
 +export PDSH_RCMD_TYPE=ssh
  
 prolog_create_key.sh prolog_create_key.sh
Line 82: Line 108:
 . vsc_start_spark.sh . vsc_start_spark.sh
  
-export SPARK_EXAMPLES=${HOME}/BigData/SparkDistributionExamples/+spark-submit --master yarn --deploy-mode client --num-executors 140 \ 
 +      --executor-memory 2G $HOME/pi.py 1000
  
-spark-submit --master yarn --deploy-mode client --num-executors 140 --executor-memory 2G $SPARK_EXAMPLES/src/main/python/pi.py 1000+. vsc_stop_spark.sh 
 +. vsc_stop_hadoop.sh
  
 epilog_discard_key.sh epilog_discard_key.sh
Line 90: Line 118:
  
 In this slurm script we have  In this slurm script we have 
-  * slurm commands, starting with #SBATCH to set the job name, the maximum execution time and where the output files will go,+  * slurm commands, starting with #SBATCH to set the job name, the maximum execution time and where the output files will go, as well as the slurm qos and partition that should be used,
   * module commands, loading all the modules which are required by our job: python, Java (Hadoop is written in Java), Hadoop, and Spark,   * module commands, loading all the modules which are required by our job: python, Java (Hadoop is written in Java), Hadoop, and Spark,
   * setup scripts which create temporary ssh keys, and start Hadoop and Spark services in user context on the nodes of the job,   * setup scripts which create temporary ssh keys, and start Hadoop and Spark services in user context on the nodes of the job,
Line 96: Line 124:
   * a script to discard the temporary ssh keys.   * a script to discard the temporary ssh keys.
  
-=== Background ===+The script is submitted to the slurm scheduler by executing: 
 +<code> 
 +$ sbatch pi.slrm 
 +</code> 
 + 
 +==== MapReduce: Concepts and example ==== 
 + 
 +MapReduce splits work into the phases 
 +  * map: a subset of the input data is processed by a single process, e.g. filtering, 
 +  * shuffle: sort the output of the map phase by its key, 
 +  * reduce: combine the sorted data 
 + 
 +An example is  
 + 
 +<code bash> 
 +#!/bin/bash 
 +#SBATCH --job-name=simple_mapreduce 
 +#SBATCH --nodes=1 --time=00:10:00 
 +#SBATCH --error=simple_mapreduce_err 
 +#SBATCH --output=simple_mapreduce_out 
 + 
 +module purge 
 +module load openjdk 
 +module load hadoop 
 + 
 +export PDSH_RCMD_TYPE=ssh 
 + 
 +prolog_create_key.sh 
 +. vsc_start_hadoop.sh 
 + 
 +hdfs dfs -mkdir input 
 +hdfs dfs -put data/wiki\_sample\_2400lines input/ 
 +hdfs dfs -rm -r tmp_out 
 + 
 +mapred streaming -D mapreduce.job.maps=4 -input input/wiki_sample_2400lines \ 
 +    -output tmp_out -mapper /bin/cat -reducer '/bin/wc -l' 
 + 
 +# check output in simple_mapreduce_out and simple_mapreduce_err 
 +# check job output on HDFS 
 +hdfs dfs -ls tmp_out 
 +echo "The number of lines is:" 
 +hdfs dfs -cat tmp_out/part-* 
 + 
 +epilog_discard_key.sh 
 +</code> 
 + 
 + 
 +==== Further examples ==== 
 + 
 +For examples in Scala, R, and SQL, including slurm scripts to run on VSC, we want to refer to the course material in [[ https://vsc.ac.at/training/2021/BigData-Mar/ | Big Data on VSC]]. 
 + 
 +==== Available modules ==== 
 + 
 +To use Big Data on VSC use those modules: 
 + 
 +  * python (note: the current spark version does NOT work with python 3.11) 
 +    * python/3.10.7-gcc-12.2.0-5a2kkeu 
 +    * python/3.9.13-gcc-12.2.0-ctxezzj 
 +    * python/3.8.12-gcc-12.2.0-tr7w5qy 
 +  * openjdk/11.0.15_10-gcc-11.3.0-ih4nksq 
 +  * hadoop (hadoop/3.3.2-gcc-11.3.0-ypaxzwt) 
 +  * spark (spark/3.1.1-gcc-11.3.0-47oqksq) 
 +  * r 
 + 
 + 
 +Depending on the application, choose the right combination: 
 +  * MapReduce => openjdk + hadoop 
 +  * Scala => openjdk + hadoop + spark 
 +  * Java => openjdk + hadoop + spark 
 +  * PySpark => openjdk + hadoop + spark + python 
 +  * R => openjdk + hadoop + spark + r 
 + 
 +==== Background ====
  
 What is VSC actually doing to run a Big Data job using Hadoop and/or Spark? What is VSC actually doing to run a Big Data job using Hadoop and/or Spark?
 +
 +== HPC ==
  
 User jobs on VSC are submitted using Slurm, thus making sure that a job gets a well defined execution environment which is similar in terms of resources and performance for each similar job execution. User jobs on VSC are submitted using Slurm, thus making sure that a job gets a well defined execution environment which is similar in terms of resources and performance for each similar job execution.
  
-Big Data jobs are often executed on Big Data clusters, which make sure that +== Big Data == 
 + 
 +Big Data jobs on the other hand are often executed on Big Data clusters, which make sure that 
   * programs are started on the nodes where data is stored, thus reducing communication overhead,   * programs are started on the nodes where data is stored, thus reducing communication overhead,
   * load balancing is done automatically,   * load balancing is done automatically,
Line 109: Line 213:
 Much of this work is done by Yarn (Yet another resource negotiator), which is a scheduler (and runs usually at a similar level on Big Data clusters as Slurm is running on VSC). Much of this work is done by Yarn (Yet another resource negotiator), which is a scheduler (and runs usually at a similar level on Big Data clusters as Slurm is running on VSC).
  
-In our setup on VSC we combine those two worlds of HPC and Big data by starting Yarn and other services in user context within a job on the nodes belonging to the job. The other services are HDFS (Hadoop distributed file system) and Spark.+== Combination of HPC and Big Data == 
 + 
 +In our setup on VSC we combine those two worlds of HPC and Big data by starting Yarn, HDFS and spark in user context within a job (on the nodes belonging to the job).
 User data can be accessed  User data can be accessed 
   * from the standard VSC paths using Spectrum Scale (formerly GPFS), or   * from the standard VSC paths using Spectrum Scale (formerly GPFS), or
-  * from HDFS, which is started locally on the local SSDs of the compute nodes. In this case the data has to be copied in and/or out before using it.+  * from HDFS, which is started locally on the local SSDs of the compute nodes. In this case the data has to be copied in and/or out at the beginning/end of the job. After a job run, the local HDFS instance is destroyed and remaining data is lost.
  
-VSC has several scripts, most are optional, which are added to the path using '''module load hadoop''' and '''module load spark''':+VSC has several scripts - most are optional - which are added to the path using '''module load hadoop''' and '''module load spark''':
   * prolog_create_key.sh and epilog_discard_key.sh, which are required to start services on other nodes, but also to access the local (master) node.   * prolog_create_key.sh and epilog_discard_key.sh, which are required to start services on other nodes, but also to access the local (master) node.
-  * vsc_start_hadoop.sh, which starts Yarn (scheduler) and HDFS (distributed file system) on the nodes of the job. (This is optional, since Spark can also run with its own scheduler and without HDFS.) +  * vsc_start_hadoop.sh, which starts Yarn (scheduler) and HDFS (distributed file system) on the nodes of the job. (If only one node is used this is optional, since Spark can also run with its own scheduler and without HDFS.)
-  * vsc_stop_hadoop.sh, which is usually not required since the services are stopped at the end by the system anyway.+
   * vsc_start_spark.sh, which starts the Spark service on the nodes of the job. (This is optional, since MapReduce jobs run without Spark.)   * vsc_start_spark.sh, which starts the Spark service on the nodes of the job. (This is optional, since MapReduce jobs run without Spark.)
 +  * vsc_stop_hadoop.sh, which is usually not required since the services are stopped at the end by the system anyway.
   * vsc_stop_spark.sh, which is usually not required since the services are stopped at the end by the system anyway.   * vsc_stop_spark.sh, which is usually not required since the services are stopped at the end by the system anyway.
  
- +==== References ====
-=== References ===+
  
  
Line 130: Line 235:
   * {{ :projects:hadoop:332-tutorial-hadoop-ahpc18.pdf |Hadoop for HPC Users: Overview and first steps}}: Tutorial at Austrian HPC Meeting 2018 by Elmar Kiesling   * {{ :projects:hadoop:332-tutorial-hadoop-ahpc18.pdf |Hadoop for HPC Users: Overview and first steps}}: Tutorial at Austrian HPC Meeting 2018 by Elmar Kiesling
  
-=== Hadoop Ecosystem ===+==== Hadoop at the TU Wien ====
  
 At the Technische Universität Wien there is a [[doku:lbd | Big Data cluster]], mainly for teaching purposes, which is also used by researchers. At the Technische Universität Wien there is a [[doku:lbd | Big Data cluster]], mainly for teaching purposes, which is also used by researchers.
    
  • doku/bigdata.1622107187.txt.gz
  • Last modified: 2021/05/27 09:19
  • by dieter