A Spark application is a set of processes running on a cluster, all these processes are coordinated by the driver program. The driver program is 1) the process where the Those processes that run computations and store data for your application are executors. Executors are 1) returning computed results to the driver program, 2) providing in-memory storage for cached RDDs\/DataFrames.<\/p>\n Execution of a Spark program:<\/p>\n \u540d\u8a5e\u89e3\u91cb\uff1a<\/p>\n ref: RDDs are divided into partitions: each partition can be considered as an immutable subset of the entire RDD. When you execute your Spark program, each partition gets sent to a worker.<\/p>\n \u5c0d RDD \u7684\u64cd\u4f5c\u53ef\u4ee5\u5206\u70ba\u5169\u7a2e\uff1a<\/p>\n \u4ee5 \u4ee5 ref: \u4ee5 \u4ee5 By default, RDDs are recomputed each time you run an action on them! Spark allows you to control what should be cached in memory.<\/p>\n ref: Moving data from one node to another across network is called "shuffling". A shuffle can occur when the resulting RDD depends on other elements from the same RDD or another RDD, you can also figure out whether a shuffle has been planned via 1)the return type of certain transformations (e.g. Operations that might cause a shuffle:<\/p>\n Narrow dependencies:<\/p>\n Wide dependencies:<\/p>\n ref: Partitions never span multiple machines, data in the same partition are guaranteed to be on the same machine. The default number of partitions is the total number of cores on all executor nodes.<\/p>\n Customizing partitions is only possible when working with Pair RDD, because of partitioning is done based on keys. The most important thing is that you must Following operations hold to (and propagate) a partitioner:<\/p>\n All other operations (e.g. ref: You can chain a call to .repartition(n) after reading the text file to specify a different and larger number of partitions. You might set this higher to match the number of cores in your cluster, for example. ref: It's really important to repartition your dataset if you are going to cache it and use for queries. The optimal number of partitions is around 4-6 for each executor core, with 40 nodes and 6 executor cores we use 1000 partitions for best performance.<\/p>\n ref: Dataset \u7684 transformation API \u5206\u6210 untyped \u548c typed\uff0c\u4f7f\u7528\u4e86 untyped API \u4e4b\u5f8c\u6703\u56de\u50b3 DataFrame\uff0c\u6703\u5931\u53bb type\uff1btyped API \u5247\u6703\u56de\u50b3 Dataset\u3002\u5982\u679c\u662f\u50cf Untyped transformations:<\/p>\n Typed transformations:<\/p>\n Datasets don't get all of the optimization the DataFrames get!<\/p>\n \u76e1\u91cf\u4f7f\u7528 Spark SQL \u7684\u5beb\u6cd5\uff08\u5373 relational operations\uff09\uff0c\u4f8b\u5982 ref: Encoders are what convert your data between JVM objects and Spark SQL's specialized internal representation. They're required by all Datasets.<\/p>\n Two ways to introduce Encoders:<\/p>\n ref: \u67d0\u500b DataFrame \u5c0f\u5230\u4e00\u53f0\u6a5f\u5668\u53ef\u4ee5\u5403\u5f97\u4e0b\uff0c\u5c31\u53ef\u4ee5 broadcast \u5b83\u3002<\/p>\n ref: If your dataset is large, you can try repartitioning to a larger number to allow more parallelism on your job. A good indication of this is if in the Spark UI \u2013 you don\u2019t have a lot of tasks, but each task is very slow to complete.<\/p>\n ref: ref: When submitting a Spark application via YARN and Spark standalone only:<\/p>\n YARN only:<\/p>\n \u8a08\u7b97\u65b9\u5f0f\uff1a<\/p>\n driver: 2cores 7.5GB x 1 ref: A Spark application is a set of processes running on a cluster, all these processes are coordinated by the driver program. The driver program is 1) the process where the main() method of your program runs, 2) the process running the code that creates a SparkSession, RDDs, DataFrames, and stages up or sends off transformations and actions.<\/p>\n","protected":false},"author":1,"featured_media":455,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[112],"tags":[108,109],"class_list":["post-454","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-about-big-data","tag-apache-spark","tag-scala"],"_links":{"self":[{"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/posts\/454","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/comments?post=454"}],"version-history":[{"count":0,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/posts\/454\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/media\/455"}],"wp:attachment":[{"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/media?parent=454"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/categories?post=454"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/vinta.ws\/code\/wp-json\/wp\/v2\/tags?post=454"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}main()<\/code> method of your program runs, 2) the process running the code that creates a SparkSession, RDDs, DataFrames, and stages up or sends off transformations and actions.<\/p>\n\n
map()<\/code>) to the executors.<\/li>\n\n
\nhttps:\/\/spark.apache.org\/docs\/latest\/cluster-overview.html<\/a>
\nhttps:\/\/databricks.com\/blog\/2016\/06\/22\/apache-spark-key-terms-explained.html<\/a><\/p>\nResilient Distributed Dataset (RDD)<\/h2>\n
\n
\n
.map()<\/code>\u3001.flatMap()<\/code>\u3001.filter()<\/code><\/li>\n<\/ul>\n<\/li>\n\n
reduce()<\/code>\u3001count()<\/code>\u3001.collect()<\/code><\/li>\n<\/ul>\n<\/li>\n<\/ul>\nsomeRDD.foreach(println)<\/code> \u70ba\u4f8b\uff0cforeach<\/code> \u662f\u500b action\uff0c\u56e0\u70ba\u5b83\u56de\u50b3\u7684\u662f Unit<\/code>\uff0cprintln<\/code> \u5be6\u969b\u4e0a\u662f print \u5230 executor \u7684 stdout \u4e86\uff0c\u6240\u4ee5\u4f60\u5728 driver program \u6839\u672c\u770b\u4e0d\u5230\u3002\u9664\u975e\u4f60\u6539\u6210 someRDD.collect().foreach(println)<\/code>\u3002<\/p>\nsomeRDD.take(10)<\/code> \u70ba\u4f8b\uff0ctake(10)<\/code> \u5be6\u969b\u4e0a\u662f\u5728 executors \u4e0a\u88ab\u904b\u7b97\u51fa\u4f86\u7684\uff0c\u4f46\u662f\u6703\u628a\u90a3 10 \u7b46\u7d50\u679c\u50b3\u56de\u5230 driver program \u4e0a\u3002<\/p>\n
\nhttps:\/\/zhangyi.gitbooks.io\/spark-in-action\/content\/chapter2\/rdd.html<\/a><\/p>\nlogRDD.filter(line => line.contains("ERROR")).take(10)<\/code> \u70ba\u4f8b\uff0cfilter()<\/code> \u662f lazy \u7684\u597d\u8655\u662f\uff0cSpark \u77e5\u9053\u6700\u5f8c\u53ea\u9700\u8981 take(10)<\/code>\uff0c\u6240\u4ee5\u7576\u5b83 filter \u96c6\u6eff 10 \u500b\u7b26\u5408\u689d\u4ef6\u7684 lines \u6642\u5c31\u53ef\u4ee5\u4e0d\u7528\u7e7c\u7e8c\u57f7\u884c\u4e0b\u53bb\u4e86\uff0c\u4e0d\u9700\u8981\u5c0d\u6574\u500b dataset \u505a filter\u3002<\/p>\nlogRDD.map(_.toLowerCase).filter(_.contains("error")).count()<\/code> \u70ba\u4f8b\uff0c\u96d6\u7136 map()<\/code> \u548c filter()<\/code> \u662f\u5169\u500b\u64cd\u4f5c\uff0c\u4f46\u662f\u56e0\u70ba\u5b83\u5011\u90fd\u662f lazy evaluation\uff0c\u6240\u4ee5 Spark \u80fd\u5920\u5728 count()<\/code> \u968e\u6bb5\u5224\u65b7\uff0c\u5176\u5be6\u53ef\u4ee5\u5728\u8b80\u53d6\u6bcf\u4e00\u884c log \u7684\u6642\u5019\u540c\u6642\u505a .toLowerCase<\/code> \u548c .contains("error")<\/code>\u3002<\/p>\n
\nhttps:\/\/www.coursera.org\/learn\/scala-spark-big-data\/lecture\/0GZV7\/evaluation-in-spark-unlike-scala-collections<\/a><\/p>\nShuffling<\/h2>\n
ShuffledRDD<\/code>); 2) using toDebugString<\/code> to see its execution plan.<\/p>\n\n
\n
\n
\nhttps:\/\/www.coursera.org\/learn\/scala-spark-big-data\/lecture\/bT1YR\/shuffling-what-it-is-and-why-its-important<\/a><\/p>\nPartitioning<\/h2>\n
cache()<\/code> or persist()<\/code> your RDDs after re-partitioning.<\/p>\n\n
map()<\/code>) will produce a result without a partitioner.<\/p>\n
\nhttps:\/\/www.coursera.org\/learn\/scala-spark-big-data\/lecture\/Vkhm0\/partitioning<\/a><\/p>\n
\nmake num of partitions equal the num of cores in your cluster<\/p>\n
\nhttps:\/\/www.safaribooksonline.com\/library\/view\/learning-apache-spark\/9781785885136\/ch01s04.html<\/a>
\nhttps:\/\/umbertogriffo.gitbooks.io\/apache-spark-best-practices-and-tuning\/content\/sparksqlshufflepartitions_draft.html<\/a><\/p>\n
\nhttps:\/\/databricks.com\/blog\/2015\/10\/13\/interactive-audience-analytics-with-apache-spark-and-hyperloglog.html<\/a>
\nhttps:\/\/stackoverflow.com\/questions\/40416357\/spark-sql-difference-between-df-repartition-and-dataframewriter-partitionby<\/a><\/p>\nDataset<\/h2>\n
select()<\/code> \u9019\u6a23\u7684 API\uff0c\u53ea\u8981\u986f\u5f0f\u5730\u52a0\u4e0a type casting\uff0c\u4f8b\u5982 ds.select($"name".as[String], $"age".as[Int])<\/code>\uff0c\u56de\u50b3\u7684\u6771\u897f\u5c31\u6703\u662f Dataset \u800c\u4e0d\u662f DataFrame\u3002<\/p>\n\n
groupBy()<\/code><\/li>\n<\/ul>\n\n
filter($"city".as[String] === "Boston")<\/code> \u6216 select($"age".as[Int])<\/code> \u7b49\uff0c\u5c11\u7528 higher-order functions\uff08\u5373 functional operations\uff09\uff0c\u4f8b\u5982 filter(p => p.city == "Boston")<\/code> \u6216 map()<\/code> \u7b49\uff0cCatalyst \u5c0d\u5f8c\u8005\u7684\u512a\u5316\u4e0d\u597d\u3002<\/p>\n
\nhttps:\/\/www.coursera.org\/learn\/scala-spark-big-data\/lecture\/yrfPh\/datasets<\/a>
\nhttps:\/\/www.51zero.com\/blog\/2016\/2\/24\/type-safety-on-spark-dataframes-part-1<\/a><\/p>\nEncoders<\/h2>\n
\n
import spark.implicits._<\/code><\/li>\norg.apache.spark.sql.Encoder<\/code><\/li>\n<\/ul>\n
\nhttps:\/\/www.coursera.org\/learn\/scala-spark-big-data\/lecture\/yrfPh\/datasets<\/a><\/p>\nBroadcast<\/h2>\n
\nhttps:\/\/umbertogriffo.gitbooks.io\/apache-spark-best-practices-and-tuning\/content\/when_to_use_broadcast_variable.html<\/a><\/p>\nSpark UI<\/h2>\n
\nhttps:\/\/databricks.com\/blog\/2015\/06\/22\/understanding-your-spark-application-through-visualization.html<\/a>
\nhttps:\/\/databricks.com\/blog\/2016\/10\/18\/7-tips-to-debug-apache-spark-code-faster-with-databricks.html<\/a>
\nhttps:\/\/docs.databricks.com\/spark\/latest\/rdd-streaming\/debugging-streaming-applications.html<\/a><\/p>\nConfigurations<\/h2>\n
\nhttps:\/\/spark.apache.org\/docs\/latest\/configuration.html<\/a>
\nhttps:\/\/spark.apache.org\/docs\/latest\/tuning.html#memory-tuning<\/a><\/p>\nMy spark-defaults.conf<\/h3>\n
spark.driver.maxResultSize 2g\nspark.jars.packages com.github.fommil.netlib:all:1.1.2,com.hankcs:hanlp:portable-1.3.4,mysql:mysql-connector-java:5.1.41,org.apache.httpcomponents:httpclient:4.5.2,org.elasticsearch.client:elasticsearch-rest-high-level-client:5.6.2\nspark.kryoserializer.buffer.max 1g\nspark.serializer org.apache.spark.serializer.KryoSerializer<\/code><\/pre>\nAllocate Resources (Executors, Cores, and Memory)<\/h2>\n
spark-submit<\/code>, you may specify following options:<\/p>\n\n
--driver-memory MEM<\/code>: Memory for driver (e.g. 1000M, 2G) (Default: 1024M).<\/li>\n--executor-memory MEM<\/code>: Memory per executor (e.g. 1000M, 2G) (Default: 1G).<\/li>\n<\/ul>\n\n
--executor-cores NUM<\/code>: Number of cores per executor. (Default: 1 in YARN mode, or all available cores on the worker in standalone mode)<\/li>\n<\/ul>\n\n
--driver-cores NUM<\/code>: Number of cores used by the driver, only in cluster mode (Default: 1).<\/li>\n--num-executors NUM<\/code>: Number of executors to launch (Default: 2).<\/li>\n<\/ul>\n
\nworker: 8cores 30GB x 4<\/p>\n\n
\n
spark.driver.memory=6g<\/code><\/li>\n<\/ul>\n<\/li>\n\n
spark.executor.cores=5<\/code><\/li>\n--executor-cores 5<\/code><\/li>\n<\/ul>\n<\/li>\n\n
\n
spark.executor.instances=4<\/code><\/li>\n--num-executors 4<\/code><\/li>\n<\/ul>\n<\/li>\n\n
\n
spark.executor.memory=21g<\/code><\/li>\n<\/ul>\n<\/li>\n<\/ul>\n
\nhttps:\/\/spark.apache.org\/docs\/latest\/submitting-applications.html<\/a>
\nhttps:\/\/spark.apache.org\/docs\/latest\/configuration.html<\/a>
\nhttps:\/\/spoddutur.github.io\/spark-notes\/distribution_of_executors_cores_and_memory_for_spark_application<\/a>
\nhttps:\/\/www.slideshare.net\/cloudera\/top-5-mistakes-to-avoid-when-writing-apache-spark-applications\/21<\/a>
\nhttp:\/\/blog.cloudera.com\/blog\/2015\/03\/how-to-tune-your-apache-spark-jobs-part-2\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"