Rdd

适用范围本文涉及到的配置项主要针对 Hadoop 2.x，Spark 2.x。 MapReduce官方文档https://hadoop.apache.org/doc...左下角: mapred-default.xml 配置项举例mapreduce.job.reduce.slowstart.completedmaps当 Map Task 完成的比例达到该值后才会为 Reduce Task 申请资源mapreduce.output.fileoutputformat.compressMapReduce Job 的结果输出需要使用压缩Spark官方文档https://spark.apache.org/docs...最后有提到想要直接配置 Hadoop 项，在 Hadoop 配置项前加 spark.hadoop 即可。Custom Hadoop/Hive Configuration 配置项示例spark.dynamicAllocation.enabled是否动态分配内存（Spark 原生配置项）spark.hadoop.mapreduce.output.fileoutputformat.compressJob 的结果输出是否使用压缩（Hadoop MR 配置项）本文出自: walker snapshot

1. Spark中的RDDResilient Distributed Datasets(弹性分布式数据集)Spark中的最基本的抽象有了RDD的存在我们就可以像操作本地集合一样操作分布式的数据包含所有元素的分区的集合RDD包含了很多的分区2. RDD中的弹性RDD中的数据是可大可小的RDD的数据默认情况下存放在内存中的，但是在内存资源不足时，Spark会自动将RDD数据写入磁盘RDD有自动容错功能,当其中一个RDD中的分区的数据丢失,或者当前节点故障时,rdd会根据依赖关系重新计算该分区的数据3. RDD在Spark中的作用迭代式计算 其主要实现思想就是RDD，把所有计算的数据保存在分布式的内存中。迭代计算通常情况下都是对同一个数据集做反复的迭代计算，数据在内存中将大大提升IO操作。这也是Spark涉及的核心：内存计算 交互式计算 因为Spark是用scala语言实现的，Spark和scala能够紧密的集成，所以Spark可以完美的运用scala的解释器，使得其中的scala可以向操作本地集合对象一样轻松操作分布式数据集 4. Spark中的名词解释ClusterManager :在Standalone模式中即为Master（主节点），控制整个集群，监控Worker。在YARN模式中为资源管理器Worker:从节点，负责控制计算节点，启动Executor。在YARN模式中为NodeManager，负责计算节点的控制。Driver 运行Application的main()函数并创建SparkContextExecutor (CoarseGrainedExecutorBackend)在worker node上执行任务的组件、用于启动线程池运行任务。每个Application拥有独立的一组ExecutorsSparkContext :整个应用的上下文，控制应用的生命周期RDD :Spark中的最基本的数据抽象DAG Scheduler : 根据DAG(有向无环图)切分stage,并且生成task,以taskset的形式返回Task Schedual: 调度task,把task交给executorStage: 一个Spark作业一般包含一到多个Stage。Task :一个Stage包含一到多个Task，通过多个Task实现并行运行的功能Transformations :转换操作,Transformation是lazy的,不会马上执行,只有当调用action时才会执行Actions : 动作SparkEnv : 线程级别的上下文，存储运行时的重要组件的引用5. 创建RDD的两种方式通过并行化集合创建RDD(用于测试) val list = List("java c++ java","java java java c++")val rdd = sc.parallelize(list)通过加载hdfs中的数据创建RDD(生产环境) val rdd = sc.textFile("hdfs://uplooking01:8020/sparktest/")6. IDEA开发Spark6.1 pom依赖<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.uplooking.bigdata</groupId> <artifactId>2018-11-08-spark</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.version>2.11.8</scala.version> <spark.version>2.2.0</spark.version> <hadoop.version>2.7.5</hadoop.version> </properties> <dependencies> <!-- 导入scala的依赖 --> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <!-- 导入spark的依赖 --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark.version}</version> </dependency> <!-- 指定hadoop-client API的版本 --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> </dependency> </dependencies> <build> <plugins> <!--编译Scala--> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> <executions> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>add-source</goal> <goal>compile</goal> </goals> </execution> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <!--编译Java--> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <executions> <execution> <phase>compile</phase> <goals> <goal>compile</goal> </goals> </execution> </executions> </plugin> <!-- 打jar插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.4.3</version> <configuration> <createDependencyReducedPom>false</createDependencyReducedPom> </configuration> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin> </plugins> </build></project>6.2 编写spark程序val conf = new SparkConf()conf.setAppName("Ops1")val sc = new SparkContext(conf)val rdd1: RDD[String] = sc.parallelize(List("java c+ java", "java java c++"))val ret = rdd1.collect().toBufferprintln(ret)6.3 打包6.4 在Driver上运行jar包spark-submit --master spark://uplooking01:7077 --class com.uplooking.bigdata.spark01.Ops1 original-spark-1.0-SNAPSHOT.jar7. 本地运行Spark程序import org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutableobject Ops1 { def main(args: Array[String]): Unit = { val conf = new SparkConf() conf.setAppName("Ops1") conf.setMaster("local[4]") val sc = new SparkContext(conf) //一般不会指定最小分区数 val rdd1 = sc.textFile("hdfs://uplooking01:8020/sparktest/") val rdd2: RDD[String] = rdd1.flatMap(line => line.split(" ")) val rdd3: RDD[(String, Int)] = rdd2.map(word => (word, 1)) val rdd4: RDD[(String, Int)] = rdd3.reduceByKey(_ + _) val ret: mutable.Buffer[(String, Int)] = rdd4.collect().toBuffer println(ret) println(rdd1.partitions.length) }}8. RDD中的分区数并行化的方式指定分区数(一般会指定分区数) ...