备案控制台
探索云世界
开发者社区 开发与运维 文章 正文

Spark Shell笔记

2023-02-23 134
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Spark Shell笔记

学习感悟


(1)学习一定要敲，感觉很简单，但是也要敲一敲，不要眼高手低

(2)一定要懂函数式编程，一定，一定

(3)shell中的方法在scala写的项目中也会有对应的方法

(4)sc和spark是程序的入口，直接用


1.png

1.png1.png1.png1.png


2.png


SparkShell


启动SparkShell


./bin/spark-shell


WordCount案例


sc.textFile("hdfs://iZm5ea99qngm2v98asii1aZ:9000/README.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://iZm5ea99qngm2v98asii1aZ:9000/out")


RDD创建(Shell)


从集合中创建RDD


parallelize和makeRDD


val rdd1246 = sc.parallelize(List("a","b","c"))


rdd1246.collect


3.png


val rdd1617=sc.makeRDD(List(1,List(("a","b","c")),(2,List("d","e","f"))))


 rdd1617.collect


4.png


从外部存储创建RDD


由外部存储系统的数据集创建，包括本地文件系统，还有Hadoop支持的数据集，如HDFS，HBase


sc.textFile("hdfs://iZm5ea99qngm2v98asii1aZ:9000/README.txt")


从其他RDD转换


常用的Transformation和Action(Shell)


map(func):返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成


scala> var rdd1638  = sc.parallelize(1 to 10)
scala> rdd1638.collect
scala> rdd1638.map(_*2).collect


6.png


filter(func):返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成


scala> var rdd1643 =sc.parallelize(1 to 10)
scala> rdd1643.filter(_>5).collect


7.png


flatMap(func):类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）


注意：func 必须是将一个数据映射为0或多个输出元素


通俗点说：一个数据通过func函数产生的集合压平


val rdd3=sc.makeRDD(List("hello1_hello2_hello3","hello4_hello5"))
scala> rdd3.flatMap(_.split("_")).collect


8.png


sample(withReplacement, fraction, seed)：以指定的随机种子随机抽样出数量为 fraction 的数据，withReplacement 表示是抽 出的数据是否放回，true 为有放回的抽样， false 为无放回的抽样，seed 用于指定随机 数生成器种子。例子从 RDD 中随机且有放 回的抽出 50%的数据，随机种子值为 3（即 可能以 1 2 3 的其中一个起始值）


scala> val rdd5 = sc.makeRDD(List(1,2,3,4,5,6,7))
scala> rdd5.sample(false,0.2,3).collect


9.png


takeSample：和 Sample 的区别是：takeSample 返回的是最终的结果集合。


10.png


union(otherDataset)：对源 RDD 和参数 RDD 求并集后返回一个 新的 RDD


intersection(otherDataset)：对源 RDD 和参数 RDD 求交集后返回一个 新的 RDD


intersection(otherDataset)：对源 RDD 和参数 RDD 求交集后返回一个 新的 RDD


11.png


distinct([numTasks]))：对源 RDD 进行去重后返回一个新的 RDD. 默认情况下，只有 8 个并行任务来操作， 但是可以传入一个可选的 numTasks 参数 改变它。


rdd3 = sc.makeRDD(List(1,1,2,3,4,4,5))
rdd3.distinct(2).collect


12.png


reduceByKey(func, [numTasks])：在一个(K,V)的 RDD 上调用，返回一个 (K,V)的 RDD，使用指定的 reduce 函数， 将相同 key 的值聚合到一起，reduce 任务 的个数可以通过第二个可选的参数来设置


13.png


groupByKey：groupByKey 也是对每个 key 进行操作，但只生成 一个 sequence。


14.png


sortByKey([ascending], [numTasks])：在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口，返回一个按照 key 进行排序 的(K,V)的 RDD


15.png


sortBy(func,[ascending], [numTasks])：与 sortByKey 类似，但是更灵活,可以用 func 先对数据进行处理，按照处理后的数 据比较结果排序。


16.png


join(otherDataset, [numTasks])：在类型为(K,V)和(K,W)的 RDD 上调用，返 回一个相同 key 对应的所有元素对在一起 的(K,(V,W))的 RDD


17.png


cogroup(otherDataset, [numTasks])：在类型为(K,V)和(K,W)的 RDD 上调用，返 回一个(K,(Iterable,Iterable))类型 的 RDD


18.png


cartesian(otherDataset)：笛卡尔积


19.png


coalesce(numPartitions)：缩减分区数，用于大数据集过滤后，提高 小数据集的执行效率。


repartition(numPartitions):根据分区数，从新通过网络随机洗牌所有 数据。


glom:将每一个分区形成一个数组，形成新的 RDD 类型时 RDD[Array[T]]


20.png


subtract:计算差的一种函数去除两个 RDD 中相同的 元素，不同的 RDD 将保留下来


21.png


mapValues:针对于(K,V)形式的类型只对 V 进行操作


22.png


reduce(func):通过 func 函数聚集 RDD 中的所有元素， 这个功能必须是可交换且可并联的


23.png


collect():在驱动程序中，以数组的形式返回数据 集的所有元素


24.png


count():返回 RDD 的元素个数


first():返回 RDD 的第一个元素（类似于 take(1)）


take(n)；返回一个由数据集的前 n 个元素组成的 数组


takeOrdered(n)：返回前几个的排序


25.png


saveAsTextFile(path)：将数据集的元素以 textfile 的形式保存 到 HDFS 文件系统或者其他支持的文件 系统，对于每个元素，Spark 将会调用 toString 方法，将它装换为文件中的文 本


saveAsSequenceFile(path)：将数据集中的元素以 Hadoop sequencefile 的格式保存到指定的目录 下，可以使 HDFS 或者其他 Hadoop 支 持的文件系统。


saveAsObjectFile(path)：用于将 RDD 中的元素序列化成对象， 存储到文件中。


countByKey();针对(K,V)类型的 RDD，返回一个 (K,Int)的 map，表示每一个 key 对应的 元素个数。


26.png


数据读取与保存主要方式(Shell)


文本文件输入输出


val rdd1 =sc.textFile("hdfs://Master:9000/cbeann/README.txt")


 rdd.saveAsTextFile("hdfs://Master:9000/cbeann/README2.txt")


JSON 、CSV文件输入输出(Shell)


先通过文本文件读入，然后通过fastjson等第三方库解析字符串为自定义的类型

先将自定义的类型通过第三方库转换为字符串，在同文本文件的形式保存到RDD中


SequenceFile 文件输入输出(Shell)


SequenceFile 文件是 Hadoop 用来存储二进制形式的 key-value 对而设计的 一种平面文件(Flat File)。


 val data=sc.parallelize(List((2,"aa"),(3,"bb"),(4,"cc"),(5,"dd"),(6,"ee")))


data.saveAsSequenceFile("hdfs://Master:9000/cbeann/seq")


val sdata = sc.sequenceFile[Int,String]("hdfs://Master:9000/cbeann/seq/p*")


对象文件输入输出(Shell)


val data=sc.parallelize(List((2,"aa"),(3,"bb"),(4,"cc"),(5,"dd"),(6,"ee")))


data.saveAsObjectFile("hdfs://master01:9000/objfile")


val objrdd:RDD[(Int,String)] = sc.objectFile[(Int,String)]("hdfs://master01:9000/objfile/p*")


Spark SQL(Shell)


启动SparkShell


./bin/spark-shell


读取数据，创建DataFrame

我的hdfs上/cbeann/person.json


 {  "name": "王小二",   "age": 15}
 {  "name": "王小三",   "age": 25}
 {  "name": "王小四",   "age": 35}


 val df = spark.read.json("hdfs://Master:9000/cbeann/person.json")


df.show


27.png


将数据注册一张表，表名为 people


df.createOrReplaceTempView("people")


28.png


发送SQL


spark.sql("select * from people where age > 16").show


29.png


或者


30.png


RDD、DataFrame、DataSet之间的转化(Shell)


RDD-》DataFrame


val rdd = sc.makeRDD(List(("zhangsan",11),("lisi",13)))
rdd.toDF("name","age").show


或者 


val rdd = sc.makeRDD(List(("zhangsan",11),("lsi",12),("wanwu",16)))
case class Person(name:String, age:Int)
 val df =  rdd.map(x=>Person(x._1,x._2.toInt)).toDF


31.png


32.png


DataFrame-》RDD


val rdd1 = df.rdd


33.png


RDD-》DataSet


val rdd = sc.makeRDD(List(("zhangsan",11),("lsi",12),("wanwu",16)))


val ds = rdd.toDS


或者


val rdd = sc.makeRDD(List(("zhangsan",11),("lsi",12),("wanwu",16)))


case class Person(name:String, age:Int)


 rdd.map(x=>Person(x._1,x._2.toInt)).toDS


34.png

35.png


DataSet-》RDD


ds.rdd


36.png

DataFrame》DataSet


scala> val rdd = sc.makeRDD(List(("zhangsan",11),("lsi",12),("wanwu",16)))
scala> val df = rdd.toDF("name","age")
scala> case class Person(name:String, age:Int)
scala> val ds = df.as[Person]
scala> ds.collect


37.png


DataSet-》DataFrame


ds.toDF


38.png


SparkSQl输入输出(Shell)


val personDF= spark.read.format("json").load("hdfs://Master:9000/cbeann/person.json")


等价于 


 val personDF1= spark.read.json("hdfs://Master:9000/cbeann/person.json")


相同的用法还有parquet,csv,text,jdbc


personDF1.write.format("json").save("hdfs://Master:9000/cbeann/person")


等价于与


personDF1.write.json("hdfs://Master:9000/cbeann/person1")


相同的用法还有parquet,csv,text,jdbc

文章标签：
Java
Shell
分布式计算
SQL
fastjson
Spark
Hadoop
Scala
存储
数据库连接
关键词：
Shell spark
Shell笔记
apache spark笔记
CBeann
目录
相关文章
热烈的马
|
5月前
|
Java Shell 分布式数据库
【大数据技术Hadoop+Spark】HBase数据模型、Shell操作、Java API示例程序讲解（附源码 超详细）
【大数据技术Hadoop+Spark】HBase数据模型、Shell操作、Java API示例程序讲解（附源码 超详细）
热烈的马
84 0
热烈的马
|
5月前
|
分布式计算 Java 大数据
【大数据技术Hadoop+Spark】HDFS Shell常用命令及HDFS Java API详解及实战（超详细 附源码）
【大数据技术Hadoop+Spark】HDFS Shell常用命令及HDFS Java API详解及实战（超详细 附源码）
热烈的马
243 0
theMilkyWay`
|
6月前
|
SQL 分布式计算 HIVE
pyspark笔记（RDD,DataFrame和Spark SQL）1
pyspark笔记（RDD,DataFrame和Spark SQL）
theMilkyWay`
56 1
程序媛地瓜
|
9月前
|
Java Shell Linux
适合初学者的Shell核心知识笔记（下）
适合初学者的Shell核心知识笔记（下）
程序媛地瓜
48 0
ALex_zry
|
6月前
|
Shell
shell脚本编程笔记（一）
.在位于头部的用以供shell明白应当以何种方式启动的该脚本的命令行中。可以提供额外参数。
ALex_zry
26 0
theMilkyWay`
|
6月前
|
SQL 存储 分布式计算
pyspark笔记（RDD,DataFrame和Spark SQL）2
pyspark笔记（RDD,DataFrame和Spark SQL）
theMilkyWay`
46 2
theMilkyWay`
|
6月前
|
分布式计算 资源调度 Java
Spark笔记（pyspark）2
Spark笔记（pyspark）
theMilkyWay`
68 0
theMilkyWay`
|
6月前
|
存储 分布式计算 资源调度
Spark笔记（pyspark）1
Spark笔记（pyspark）
theMilkyWay`
70 0
开化者
|
9月前
|
机器学习/深度学习 监控 Shell
Shell 一个月时间让你从小白到实战【万字笔记建议收藏方便学习】2
Shell 一个月时间让你从小白到实战【万字笔记建议收藏方便学习】2
开化者
40 0
开化者
|
9月前
|
Shell Linux
Shell 一个月时间让你从小白到实战【万字笔记建议收藏方便学习】1
Shell 一个月时间让你从小白到实战【万字笔记建议收藏方便学习】1
开化者
54 0

热门文章

最新文章

  • 1
    shell中for while until 三种循环的用法
  • 2
    ecs服务器shell常用脚本练习（十）
  • 3
    ecs服务器shell常用脚本练习（九）
  • 4
    ecs服务器shell常用脚本练习（六）
  • 5
    ecs服务器shell常用脚本练习（二）
  • 6
    ecs服务器shell常用脚本练习（四）
  • 7
    每天解析一个shell脚本（58）
  • 8
    Linux之Shell基本命令篇
  • 9
    利用Shell脚本编写局域网监控软件：实时监测主机连接情况
  • 10
    Shell编程
  • 1
    Spark学习---SparkSQL（概述、编程、数据的加载和保存、自定义UDFA、项目实战）
    117
  • 2
    安装Spark
    35
  • 3
    Spark学习---day07、Spark内核（Shuffle、任务执行）
    42
  • 4
    Spark读取变更Hudi数据集Schema实现分析
    39
  • 5
    实战|使用Spark Streaming写入Hudi
    49
  • 6
    adb spark的lakehouse api访问内表数据，还支持算子下推吗
    107
  • 7
    Spark学习---day06、Spark内核（源码提交流程、任务执行）
    43
  • 8
    Spark学习--day05、SparkCore电商网站实操、SparkCore-工程代码
    67
  • 9
    Spark【Spark学习大纲】简介+生态+RDD+安装+使用（xmind分享）
    32
  • 10
    Spark【基础知识 03】【RDD常用算子详解】（图片来源于网络）
    60

    • 相关课程

    更多
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第二阶段
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第三阶段
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第六阶段
  • Linux Shell 编程入门与实战
  • Shell 编程入门到精通
  • 大数据实时计算框架Spark快速入门

    • 相关电子书

    更多
  • Apache Spark: Cloud and On-Prem
  • Hybrid Cloud and Apache Spark
  • Scalable Deep Learning on Spark

    • 相关实验场景

    更多
  • Alibaba Cloud Linux操作系统Shell程序
    • 下一篇
    2024年阿里云免费云服务器及学生云服务器申请教程参考