spark复习总结01

发布时间:2017-7-9 7:33:30编辑:www.fx114.net 分享查询网我要评论
本篇文章主要介绍了"spark复习总结01 ",主要涉及到spark复习总结01 方面的内容,对于spark复习总结01 感兴趣的同学可以参考一下。

2.初始化spark

SparkConf conf = new SparkConf().setAppName("LocalWordCount").setMaster("local");// 指定运行在本地JavaSparkContext sparkContext = new JavaSparkContext(conf);

  master为Spark、Mesos、YARN URL或local。使用local值时,表示在本地模式下运行程序。

  也可以在省略,在spark-submit时进行指定。

3.使用spark-shell,会自动创建sparkContext sc

  --marter 设置运行模式

$ ./bin/spark-shell --master local[4]   //4为cup核数

  --jars 将jar包添加到classpath,多个jar中间用,隔开

$ ./bin/spark-shell --master local[4] --jars code.jar

  --packages 通过maven坐标添加jar,多个用逗号隔开

$ ./bin/spark-shell --master local[4] --packages "org.example:example:0.1"

  --repositories 添加外部的repository

4.RDD的创建

  4.1 通过并行化集合创建

 //通过并行化集合来创建RDDcontext.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

  将集合并行化时可以指定分区数,spark会在每个分区上启动一个task。

  通常集群的每个CPU设置2~4个分区

  通常,spark基于集群自动设置分区数,也可以手动设置,parallelize(array,10)

  4.2 通过读取外部文件创建

// 通过读取外部文件创建RDDsparkContext.textFile("src/main/resources/wordcount.txt", 5);

  如果使用的是本地文件,在其他worker节点上该文件必须在相同的目录下

  textFile()中可以是文件和目录,同时可以使用通配符sparkContext.textFile("src/main/resources/*.txt", 5);

  分区数默认情况spark为每个块创建一个分区,也可以手动设置,但不能小于块数。

  同时读取不同目录下的文件,使用wholeTextFiles(),仅适用于小文档

  

/** * 读取不同目录下的文件,返回<fileName,content>的RDD,仅适用于小文档 *  * [(file:/D:/workhome/workhome2/beifeng-spark/src/main/resources/wordcount.txt,      hello scala java      hello spark      hello yarn      hadoop spark yarn),   (file:/D:/workhome/workhome2/beifeng-spark/src/main/resources/other/aa.txt,      hadoop yarn      scale java)]*/JavaPairRDD<String, String> files=sparkContext.wholeTextFiles("src/main/resources/wordcount.txt,src/main/resources/other/aa.txt");       //返回以每个文档为元素的RDDJavaRDD<String> textRdd=files.map(new Function<Tuple2<String,String>, String>() {            @Override            public String call(Tuple2<String, String> v1) throws Exception {                //将文档中的换行符替换为空格,将回车符去掉                return v1._2.replaceAll("\r"," ").replaceAll("\n", "");            }});        //返回以每个单词为元素的RDD JavaRDD<String> words=textRdd.flatMap(new FlatMapFunction<String, String>() {            @Override            public Iterable<String> call(String text) throws Exception {                return Arrays.asList(text.split(" "));            } });

 5.cache(),persist()和checkpoint()之间的区别

  5.1 cache()和checkpoint()的区别

    适用场景:

      cache:会被重复使用但不太大的RDD,只会使用内存

      checkpoint:运算时间过长或者运算量太大才能得到的RDD,或者依赖其他RDD过多的RDD

    缓存机制:

      cache:每计算出一个要cache的RDD就直接将其cache到内存。

      checkpoint:等到job结束后另外开启专门的job去完成checkpoint,即checkpoint的RDD会计算两次,

          因此在使用rdd.checkpoint的时候,建议加上rdd.cache(),第二次的计算的时候就会直接使用cache中的内容

  5.2 persisit()和checkpoint()的区别

    persist()可以将RDD的partition持久化到磁盘,但该partition由blockManager管理,一旦driver program执行结束,也就是executor所在的进程停止,blockManager也会停止,别cache到磁盘上的RDD也会被清空。

    checkpoint()可以将RDD持久化到HDFS或本地文件夹,如果不手动清除永久存在。

  5.3 cache()和persisit()的区别

    cache()底层调用persist(StorageLevel.MEMORY_ONLY)


上一篇:教你如何让数据库支持emoji表情符存储

相关文章

相关评论

本站评论功能暂时取消,后续此功能例行通知。

一、不得利用本站危害国家安全、泄露国家秘密,不得侵犯国家社会集体的和公民的合法权益,不得利用本站制作、复制和传播不法有害信息!

二、互相尊重,对自己的言论和行为负责。

好贷网好贷款