In-Memory Computation with Spark Lecture BigData Analytics Julian M. Kunkel julian.kunkel@googlemail.com University of Hamburg / German Climate Computing Center (DKRZ) 2017-01-20 Disclaimer: Big Data software is constantly updated, code samples may be outdated.
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Outline 1 Concepts 2 Architecture 3 Computation 4 Managing Jobs 5 Examples 6 Higher-Level Abstractions 7 Summary Julian M. Kunkel Lecture BigData Analytics, 2016 2 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary In-Memory Computation/Processing/Analytics [26] In-memory processing: Processing data stored in memory (database) Advantage: No slow I/O necessary ⇒ fast response times Disadvantages Data must fit in the memory of the distributed storage/database Additional persistency (with asynchronous flushing) usually required Fault-tolerance is mandatory BI-Solution: SAP Hana Big data approaches: Apache Spark, Apache Flink Julian M. Kunkel Lecture BigData Analytics, 2016 3 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Overview to Spark [10, 12] In-memory processing (and storage) engine Load data from HDFS, Cassandra, HBase Resource management via. YARN, Mesos, Spark, Amazon EC2 ⇒ It can use Hadoop but also works standalone! Task scheduling and monitoring Rich APIs APIs for Java, Scala, Python, R Thrift JDBC/ODBC server for SQL High-level domain-specific tools/languages Advanced APIs simplify typical computation tasks Interactive shells with tight integration spark-shell: Scala (object-oriented functional language running on JVM) pyspark: Python sparkR: R (basic support) Execution in either local (single node) or cluster mode Julian M. Kunkel Lecture BigData Analytics, 2016 4 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Data Model [13] Distributed memory model: Resilient Distributed Datasets (RDDs) Named collection of elements distributed in partitions P1 P2 P3 P4 RDD X X = [1,2,3,4,5, ...,1000] distributed into 4 partitions Typically a list or a map (key-value pairs) A RDD is immutatable, e.g., cannot be changed High-level APIs provide additional representations e.g., SparkSQL uses DataFrames (aka tables) Shared variables offer shared memory access Durability of data RDDs live until the SparkContext is terminated To keep them, they need to be persisted (e.g., to HDFS) Fault-tolerance is provided by re-computing data (if an error occurs) Julian M. Kunkel Lecture BigData Analytics, 2016 5 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Resilient Distributed Datasets (RDDs) [13] Creation of a RDD by either Parallelizing an existing collection 1 data = [1, 2, 3, 4, 5] 2 rdd = sc.parallelize(data, 5) # create 5 partitions Referencing a dataset on distributed storage, HDFS, ... 1 rdd = sc.textFile("data.txt") RDDs can be transformed into derived (newly named) RDDs 1 rdd2 = rdd.filter( lambda x : (x % 2 == 0) ) # operation: filter odd tuples Computation runs in parallel on the partitions Usually re-computed, but RDD can be kept in memory or stored if large RDDs can be redistributed (called shuffle) Knows its data lineage (how it was computed) Fault-tolerant collection of elements (lists, dictionaries) Split into choosable number of partitions and distributed Derived RDDs can be re-computed by using the recorded lineage Julian M. Kunkel Lecture BigData Analytics, 2016 6 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Shared Variables [13] Broadcast variables (for read-only access): transfer to all executors For readability, do not modify the broadcast variable later 1 broadcastVar = sc.broadcast([1, 2, 3]) 2 print (broadcastVar.value) 3 # [1, 2, 3] Accumulators (reduce variables): Counters that can be incremented Other data types can be supported: 1 accum = sc.accumulator(0) # Integer accumulator 2 accum.add(4) 3 4 # Accumulator for adding vectors: 5 class VectorAccumulatorParam(AccumulatorParam): 6 def zero(self, initialValue): 7 return Vector.zeros(initialValue.size) 8 9 def addInPlace(self, v1, v2): 10 v1 += v2 11 return v1 12 # Create an accumulator 13 vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam()) Julian M. Kunkel Lecture BigData Analytics, 2016 7 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary 1 Concepts 2 Architecture 3 Computation 4 Managing Jobs 5 Examples 6 Higher-Level Abstractions 7 Summary Julian M. Kunkel Lecture BigData Analytics, 2016 8 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Execution of Applications [12, 21] Source: [12] Driver program: process runs main(), creates/uses SparkContext Task: A unit of work processed by one executor Job: A spark action triggering computation starts a job Stage: collection of tasks executing the same code; run concurrently Works independently on partitions without data shuffling Executor process: provides slots to runs tasks Isolates apps, thus data cannot be shared between apps Cluster manager: allocates cluster resources and runs executor Julian M. Kunkel Lecture BigData Analytics, 2016 9 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Data Processing [13] Driver (main program) controls data flow/computation Executor processes are spawned on nodes Store and manage RDDs Perform computation (usually on local partition) In local only one executor is created Execution of code 1 The closure is computed: variables/methods needed for execution 2 The driver serializes the closure together with the task (code) Broadcast vars are useful as they do not require to be packed with the task 3 The driver sends the closure to the executors 4 Tasks on the executor run the closure which manipulates the local data Julian M. Kunkel Lecture BigData Analytics, 2016 10 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Persistence [13] Concepts The data lineage of an RDD is stored Actions trigger computation, no intermediate results are kept The methods cache() and persist() enables preserving of results The first time a RDD is computed, it is then kept for further usage Each executor keeps its local data cache() keeps data in memory (level: MEMORY_ONLY) persist() allows to choose the storage level Spark manages the memory cache automatically LRU cache, old RDDs are evicted to secondary storage (or deleted) If an RDD is not in cache, re-computation may be triggered Storage levels MEMORY_ONLY: keep Java objects in memory, or re-compute them MEMORY_AND_DISK: keep Java objects in memory or store them on disk MEMORY_ONLY_SER: keep serialized Java objects in memory DISK_ONLY: store the data only on secondary storage Julian M. Kunkel Lecture BigData Analytics, 2016 11 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Parallelism [13] Spark runs one task for each partition of the RDD Recommendation: create 2-4 partitions for each CPU When creating a RDD a default value is set, but can be changed manually 1 # Create 10 partitions when the data is distributed 2 sc.parallelize(data, 10) The number of partitions is inherited from the parent(s) RDD Shuffle operations contain the argument numTasks Define the number of partitions for the new RDD Some actions/transformations contain numTask Define the number of reducers By default, 8 parallel tasks for groupByKey() and reduceByKey() Analyze the data partitioning using glom() It returns a list with RDD elements for each partition 1 X.glom().collect() 2 # [[], [ 1, 4 ], [ ], [ 5 ], [ 2 ] ] => here we have 5 partitions for RDD X 3 # Existing values are 1, 2, 4, 5 (not balanced in this example) Julian M. Kunkel Lecture BigData Analytics, 2016 12 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary 1 Concepts 2 Architecture 3 Computation 4 Managing Jobs 5 Examples 6 Higher-Level Abstractions 7 Summary Julian M. Kunkel Lecture BigData Analytics, 2016 13 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Computation Lazy execution: apply operations when results are needed (by actions) Intermediate RDDs can be re-computed multiple times Users can persist RDDs (in-memory or disk) for later use Many operations apply user-defined functions or lambda expressions Code and closure are serialized on the driver and send to executors Note: When using class instance functions, the object is serialized RDD partitions are processed in parallel (data parallelism) Use local data where possible RDD Operations [13] Transformations: create a new RDD locally by applying operations Actions: return values to the driver program Shuffle operations: re-distribute data across executors Julian M. Kunkel Lecture BigData Analytics, 2016 14 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Simple Example Example session when using pyspark To run with a specific Python version, e.g., use 1 PYSPARK_PYTHON=python3 pyspark --master yarn-client Example data-intensive python program 1 # Distribute the data: here we have a list of numbers from 1 to 10 million 2 # Store the data in an RDD called nums 3 nums = sc.parallelize( range(1,10000000) ) 4 5 # Compute a derived RDD by filtering odd values 6 r1 = nums.filter( lambda x : (x % 2 == 1) ) 7 8 # Now compute squares for all remaining values and store key/value tuples 9 result = r1.map( lambda x : (x, x*x*x) ) 10 11 # Retrieve all distributed values into the driver and print them 12 # This will actually run the computation 13 print(result.collect()) 14 # [(1, 1), (3, 27), (5, 125), (7, 343), (9, 729), (11, 1331), ... ] 15 16 # Store results in memory 17 resultCached = result.cache() Julian M. Kunkel Lecture BigData Analytics, 2016 15 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Compute PI [20] Approach: Randomly throw NUM_SAMPLES darts on a circle and count hits Python 1 def sample(p): 2 x, y = random(), random() 3 return 1 if x*x + y*y < 1 else 0 4 5 count = sc.parallelize(xrange(0, NUM_SAMPLES)).map(sample).reduce(lambda a, b: a + b) 6 print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES) Java 1 int count = spark.parallelize(makeRange(1, NUM_SAMPLES)).filter( 2 new Function<Integer, Boolean>() { 3 public Boolean call(Integer i) { 4 double x = Math.random(); 5 double y = Math.random(); 6 return x*x + y*y < 1; 7 } 8 }).count(); 9 System.out.println("Pi is roughly " + 4 * count / NUM_SAMPLES); Julian M. Kunkel Lecture BigData Analytics, 2016 16 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Transformations Create a New RDD [13] All RDDs support map(func): pass each element through func filter(func): include those elements for which func returns true flatMap(func): similar to map, but func returns a list of elements mapPartitions(func): like map but runs on each partition independently sample(withReplacement, fraction, seed): pick a random fraction of data union(otherDataset): combine two datasets intersection(otherDataset): set that contains only elements in both sets distinct([numTasks]): returns unique elements cartesian(otherDataset): returns all pairs of elements pipe(command, [envVars]): pipe all partitions through a program Remember: Transformations return a lazy reference to a new dataset Julian M. Kunkel Lecture BigData Analytics, 2016 17 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Transformations Create a New RDD [13] Key/Value RDDs additionally support groupByKey([numTasks]): combines values of identical keys in a list reduceByKey(func, [numTasks]): aggregate all values of each key aggregateByKey(zeroValue, seqOp, combOp, [numTasks]): aggregates values for keys, uses neutral element sortByKey([ascending], [numTasks]): order the dataset join(otherDataset, [numTasks]): pairs (K,V) elements with (K,U) and returns (K, (V,U)) cogroup(otherDataset, [numTasks]): returns (K, iterableV, iterableU) Julian M. Kunkel Lecture BigData Analytics, 2016 18 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Actions: Perform I/O or return data to the driver [13] reduce(func): aggregates elements func(x,y) ⇒z Func should be commutative and associative count(): number of RDD elements countByKey(): for K/V, returns hashmap with count for each key foreach(func): run the function on each element of the dataset Useful to update an accumulator or interact with storage collect(): returns the complete dataset to the driver first(): first element of the dataset take(n): array with the first n elements takeSample(withReplacement, num, [seed]): return random array takeOrdered(n, [comparator]): first elements according to an order saveAsTextFile(path): convert elements to string and write to a file saveAsSequenceFile(path): ... saveAsObjectFile(path): uses Java serialization Julian M. Kunkel Lecture BigData Analytics, 2016 19 / 53
Concepts Architecture Computation Managing Jobs Examples Higher-Level Abstractions Summary Shuffle [13] Concepts Repartitions the RDD across the executors Costly operation (requires all-to-all) May be triggered implicitly by operations or can be enforced Requires network communication The number of partitions can be set Operations repartition(numPartitions): reshuffle the data randomly into partitions coalesce(numPartitions): decrease the number of partitions1 repartionAndSortWithinPartitions(partitioner): repartition according to the partitioner, then sort each partition locally1 1More efficient than repartition() Julian M. Kunkel Lecture BigData Analytics, 2016 20 / 53
Hiển thị 20 trang đầu. Tải tài liệu để xem đầy đủ.
Chưa có bình luận nào. Hãy là người đầu tiên!
Bình luận (0)
Chưa có bình luận nào. Hãy là người đầu tiên!