Spark第三课

这篇具有很好参考价值的文章主要介绍了Spark第三课。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.分区规则

1.分区规则

shuffle
1.打乱顺序
2.重新组合

1.分区的规则

默认与MapReduce的规则一致,都是按照哈希值取余进行分配.
一个分区可以多个组,一个组的数据必须一个分区

2. 分组的分区导致数据倾斜怎么解决?

  • 扩容 让分区变多
  • 修改分区规则

3.HashMap扩容为什么必须是2的倍数?

当不是2的倍数时, 好多的位置取不到
比如 为5 01234 123都取不到
必须保证,相关的位数全是1,所以必定2的倍数 2的n次方
所以位运算不是什么时候都能用的
Spark第三课,Spark,大数据,spark

2.转换算子

1.单值转换算子

1.filter过滤器

1.注意

过滤只是将数据进行校验,而不是修改数据. 结果为true就保留,false就丢弃
Spark第三课,Spark,大数据,spark

2.代码
JavaSparkContext sc = new JavaSparkContext("local[*]","filter");

List<String> dataList = Arrays.asList("giao","giao2","zhangsan","lisi");
JavaRDD<String> rdd1 = sc.parallelize(dataList);
//JavaRDD<String> rddFilter1 = rdd1.filter(null);
JavaRDD<String>  rddFilter2= rdd1.filter(s->s.substring(0,1).toLowerCase().equals("g"));
//rddFilter1.collect().forEach(System.out::println);
System.out.println("----------------------------");
rddFilter2.collect().forEach(System.out::println);

Spark第三课,Spark,大数据,spark

2.dinstinct

1.原理

分组
通过使用分组取重,相同的话,都是一个组了,所以Key唯一
应该是先分组,然后吧K提出来就好了

2.代码
JavaSparkContext sc = new JavaSparkContext("local[*]","Distinct");

List<String> dataList = Arrays.asList("giao1","gg1","gg1","gg2","gg2","gg1","gg3","gg1","gg5","gg3");
JavaRDD<String> rdd1 = sc.parallelize(dataList);
JavaRDD<String> rddDistinct = rdd1.distinct();
rddDistinct.collect().forEach(System.out::println);

Spark第三课,Spark,大数据,spark

3.排序

1.介绍

sortby方法需要传3个参数
参数1 排序规则
参数2 升序还是降序(false) 默认升序(true)
参数3 排序的分区数量(说明方法底层是靠shuffle实现,所以才有改变分区的能力)

2.排序规则

排序规则,是按照结果去排序
其实是用结果生成一个K值,通过K值进行排序,然后展示 V值
或者说权值, 按照权值排序
将Value变成K V

3.代码
 public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local[*]","SparkSort");
        List<String> dataList = Arrays.asList("kunkun","giaogiao","GSD","JJ","chenzhen","Lixiaolong");
        JavaRDD<String> rdd1 = sc.parallelize(dataList);
        JavaRDD<String> rddSort = rdd1.sortBy(s -> {
            switch (s.substring(0, 1).toLowerCase()) {
                case "k":
                    return 5;
                case "g":
                    return 3;
                case "j":
                    return 1;
                case "c":
                    return 2;
                case "l":
                    return 4;

            }
            return null;
        }, false, 3);
        rddSort.collect().forEach(System.out::println);

    }

2.键值对转换算子

1.介绍

1.什么是键值对转换算子

如何区分是键值对方法还是单值方法呢?
通过参数来判断, 如果参数是一个值,就是单值,如果是2个,就是键值对

2.元组是不是键值对?
public static void main(String[] args) {
    JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");
    List<Integer> dataList = Arrays.asList(1, 2, 3, 4, 5);
    JavaRDD<Integer> rdd1 = sc.parallelize(dataList);
    JavaRDD<Tuple2> rddmap = rdd1.map(num -> new Tuple2(num, num));
    rddmap.collect().forEach(System.out::println);
}

Spark第三课,Spark,大数据,spark
答案是,不是,因为这个的返回值,是一个元组,而元组整体,是一个单值,所以,是单值
只有返回值 是RDD<K1,V1 >的时候,才是键值对类型算子

3. 使用Pair转换键值对算子
public static void main(String[] args) {
    JavaSparkContext sc  = new JavaSparkContext("local[*]","RddPair");
    List<Integer> dataList = Arrays.asList(1, 2, 3, 4, 5);
    JavaRDD<Integer> rdd = sc.parallelize(dataList);
    JavaPairRDD<Integer, Integer> rddPair = rdd.mapToPair(num -> new Tuple2<>(num, num));
    rddPair.collect().forEach(System.out::println);

}

Spark第三课,Spark,大数据,spark

4.直接在获取时转换键值对

这里使用的是parallelizePairs方法 获取的是JavaPairRDD

public static void main(String[] args) {
    JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");
    JavaPairRDD<String, Integer> rddPair = sc.parallelizePairs(Arrays.asList(
            new Tuple2<>("a", 1),
            new Tuple2<>("a", 2),
            new Tuple2<>("b", 1),
            new Tuple2<>("b", 1),
            new Tuple2<>("c", 2),
            new Tuple2<>("c", 1)
    ));
    rddPair.collect().forEach(System.out::println);

}

Spark第三课,Spark,大数据,spark

5.分组来获取键值对

```java
public static void main(String[] args) {
    JavaSparkContext sc  = new JavaSparkContext("local[*]","RddPair");
    List<String> dataList = Arrays.asList("aa","bb","aa","bb","cc");
    JavaRDD<String> rdd = sc.parallelize(dataList);
    JavaPairRDD<Object, Iterable<String>> rddGroup = rdd.groupBy(s->s);
    rddGroup.collect().forEach(System.out::println);
}

Spark第三课,Spark,大数据,spark

2.mapValue方法

1.介绍

直接对value进行操作,不需要管K
当然,也有mapKey方法可以无视Value操作Key

2.代码演示
  public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");
        JavaPairRDD<String, Integer> rddPair = sc.parallelizePairs(Arrays.asList(
                new Tuple2<>("a", 1),
                new Tuple2<>("a", 2),
                new Tuple2<>("b", 1),
                new Tuple2<>("b", 1),
                new Tuple2<>("c", 2),
                new Tuple2<>("c", 1)
        ));
        JavaPairRDD<String, Integer> mapV = rddPair.mapValues(num -> num * 2);
        mapV.collect().forEach(System.out::println);

    }

Spark第三课,Spark,大数据,spark

3.WordCount实现

iter.spliterator().estimateSize());
spliterator
Spliterator(Split Iterator)是Java 8引入的一个新接口,用于支持并行遍历和操作数据。它是Iterator的扩展,可以用于在并行流(Parallel Stream)中对数据进行划分和遍历,从而实现更高效的并行处理
spliterator()方法是在Iterable接口中定义的一个默认方法,用于生成一个Spliterator对象,以支持数据的并行遍历。它的具体作用是将Iterable中的数据转换为一个可以在并行流中使用的Spliterator对象。

estimateSize

estimateSize()方法是Java中Spliterator接口的一个方法,用于估算Spliterator所包含的元素数量的大小。Spliterator是用于支持并行遍历和操作数据的接口,而estimateSize()方法提供了一个估计值,用于在处理数据时预测Spliterator包含的元素数量。

public static void main(String[] args) {
    JavaSparkContext sc  = new JavaSparkContext("local[*]","RddPair");
    List<String> dataList = Arrays.asList("aa","bb","aa","bb","cc");
    JavaRDD<String> rdd = sc.parallelize(dataList);
    JavaPairRDD<Object, Iterable<String>> rddGroup = rdd.groupBy(s->s);
    JavaPairRDD<Object, Long> wordCount = rddGroup.mapValues(iter -> iter.spliterator().estimateSize());
    wordCount.collect().forEach(System.out::println);
}

Spark第三课,Spark,大数据,spark

3.groupby 与groupByKey

1 .代码
public static void main(String[] args) {
    JavaSparkContext sc = new JavaSparkContext("local[*]","G1");
    JavaPairRDD<String, Integer> rddPair;
    rddPair = sc.parallelizePairs(Arrays.asList(
            new Tuple2<>("a", 1),
            new Tuple2<>("a", 2),
            new Tuple2<>("b", 1),
            new Tuple2<>("b", 1),
            new Tuple2<>("c", 2),
            new Tuple2<>("c", 1)
    ));
    JavaPairRDD<String, Iterable<Integer>> rddGroupByKey = rddPair.groupByKey();
    JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> rddGroupBy = rddPair.groupBy(t -> t._1);
    rddGroupByKey.collect().forEach(System.out::println);

}

Spark第三课,Spark,大数据,spark

2.分析区别
  • 1.参数
    GroupBy是自选规则 而GroupByKey是将PairRDD的Key当做分组规则
  • 2.结果
    GroupBy是将作为单值去分组,即使RDD是Pair, 而GroupByKey 则是将K V分开 ,将V作为组成员
3.注意

GroupByKey是不能进行随意使用的,底层用的含有shuffle,如果计算平均值,就不能通过GroupByKey直接进行计算.

4.reduce与reduceByKey

1.介绍

多个变量进行同样的运算规则
Stream是1.8新特性,
计算的本质 两两结合
Spark第三课,Spark,大数据,spark
reduce

2. 代码
  public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local[*]","Reduce");
        JavaPairRDD<String, Integer> rddPair;
        rddPair = sc.parallelizePairs(Arrays.asList(
                new Tuple2<>("a", 1),
                new Tuple2<>("a", 2),
                new Tuple2<>("b", 1),
                new Tuple2<>("b", 1),
                new Tuple2<>("c", 2),
                new Tuple2<>("c", 1)
        ));
        
        rddPair.reduceByKey(Integer::sum).collect().forEach(System.out::println);
    }

Spark第三课,Spark,大数据,spark

3.理解

相同Key值的V进行运算,所以底层是有分组的,所以底层是一定有Shuffle,一定有改变分区的能力,改变分区数量和分区规则.

4.与groupByKey区别

reduceByKey
将相同key的数量中1的V进行两两聚合
Spark第三课,Spark,大数据,spark
reduceByKey 相同的key两两聚合,在shuffle落盘之前对分区内数据进行聚合,这样会减少落盘数据量,并不会影响最终结果(预聚合) 这就是combine
Spark第三课,Spark,大数据,spark

有钱先整IBM小型机

Shuffle优化
1.花钱
2.调大缓冲区(溢出次数减少)
3.

sortByKey
想比较必须实现可比较的接口
默认排序规则为升序,
通过K对键值对进行排序

行动算子
通过调用RDD方法让Spark的功能行动起来
Spark第三课,Spark,大数据,spark
map 是在new
Spark第三课,Spark,大数据,spark

转换算子 得到的是RDD
注意 转换跑不起来 行动能跑起来 这句话是错误的

当使用sort时,也是能跑起来的,但是还是转换算子
Spark第三课,Spark,大数据,spark
第一行运行占用内存,第一个for 运算需要内存,但是第一行占用了大量内存,所以第一行浪费了,这就需要懒加载,所以第一行的执行时机是在第二个for运行前使用的.

注意map collect 不是懒加载,只是没人调用他的job(RDD算子内部的代码)
RDD算子外部的代码都是在Driver端文章来源地址https://www.toymoban.com/news/detail-680876.html

到了这里,关于Spark第三课的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • MyBatis第三课

    目录 回顾  #和$区别 #(预编译SQL)和$(即时SQL,它是进行的字符串拼接)的区别,其中之一就是预编译SQL和即时SQL的区别 原因: 两者的共同点 MaBits可以看作是Java程序和Mysql的沟通桥梁,底层还是(jdbc) 访问数据库还得是Mysql 多表查询(慢) 1.通常情况下,数据库集群是很多

    2024年01月20日
    浏览(37)
  • 第三课:GPT

    GPT出现的原因 未标注的文本数据远多于已标注的文本数据,并且对于不同的下游任务会存在不同的标注方式 GPT的方法原理 半监督学习 基于大量未标注的文本数据,训练预训练语言模型 使用已标注文本数据,对模型针对某一特定下游任务进行finetune,只更改output layer(线性层

    2024年01月21日
    浏览(43)
  • Kafka第三课

    Flume 由三部分 Source Channel Sink 可以通过配置拦截器和Channel选择器,来实现对数据的分流, 可以通过对channel的2个存储容量的的设置,来实现对流速的控制 Kafka 同样由三大部分组成 生产者 服务器 消费者 生产者负责发送数据给服务器 服务器存储数据 消费者通过从服务器取数据 但

    2024年02月13日
    浏览(33)
  • springboot综合案例第三课

    Spring Security 的前身是 Acegi Security ,是 Spring 项目组中用来提供安全认证服务的框架。 (https://projects.spring.io/spring-security/) Spring Security 为基于J2EE企业应用软件提供了全面安全服务。特别 是使用领先的J2EE解决方案-Spring框架开发的企业软件项目。人们使用Spring Security有很多种原

    2024年02月12日
    浏览(31)
  • 大数据技术之Spark——Spark SQL

            Spark SQL是Spark用于结构化数据处理的Spark模块。         我们之前学习过hive,hive是一个基于hadoop的SQL引擎工具,目的是为了简化mapreduce的开发。由于mapreduce开发效率不高,且学习较为困难,为了提高mapreduce的开发效率,出现了hive,用SQL的方式来简化mapreduce:hiv

    2024年02月12日
    浏览(53)
  • 大数据技术之Spark(一)——Spark概述

    大数据技术之Spark(一)——Spark概述 Apache Spark是一个开源的、强大的分布式 查询和处理引擎 ,它提供MapReduce的灵活性和可扩展性,但速度明显要快上很多;拿数据存储在内存中的时候来说,它比Apache Hadoop 快100倍,访问磁盘时也要快上10倍。 Spark 是一种由 Scala 语言开发的快

    2024年02月14日
    浏览(34)
  • 【大数据开发 Spark】第一篇:Spark 简介、Spark 的核心组成(5大模块)、Spark 的主要特征(4大特征)、Spark 对比 MapReduce

    初步了解一项技术,最好的方式就是去它的官网首页,一般首页都会有十分官方且准确的介绍,学习 Spark 也不例外, 官方介绍:Apache Spark ™是一种多语言引擎,用于在单节点机器或集群上执行数据工程、数据科学和机器学习。 我们可以得知,Spark 可以单节点运行,也可以搭

    2024年02月05日
    浏览(48)
  • 【spark大数据】spark大数据处理技术入门项目--购物信息分析

    购物信息分析基于spark 目录 本案例中三个文案例中需要处理的文件为 order_goods.txt、products.txt 以及 orders.txt 三个文件,三个文件的说明如下 一、本实训项目针对实验数据主要完成了哪些处理? 二、Hadoop+Spark集群环境的搭建步骤有哪些?(只介绍完全分布式集群环境的搭建)

    2023年04月08日
    浏览(67)
  • Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】

    Saprk 提供了两种方法来实现从 RDD 转换得到 DataFrame: 利用反射机制推断 RDD 模式 使用编程方式定义 RDD 模式 下面使用到的数据 people.txt :         在利用反射机制推断 RDD 模式的过程时,需要先定义一个 case 类,因为只有 case 类才能被 Spark 隐式地转换为DataFrame对象。 注意

    2024年02月09日
    浏览(55)
  • [机器学习、Spark]Spark MLlib实现数据基本统计

    👨‍🎓👨‍🎓博主:发量不足 📑📑本期更新内容: Spark MLlib基本统计 📑📑下篇文章预告:Spark MLlib的分类🔥🔥 简介:耐心,自信来源于你强大的思想和知识基础!!   目录 Spark MLlib基本统计 一.摘要统计 二.相关统计 三.分层抽样   MLlib提供了很多统计方法,包含

    2024年02月02日
    浏览(48)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包