0


spark方法

spark

1,Spark概述

Spark是一个开源的、大规模的、快速的数据处理引擎,专为大规模数据处理而设计。它提供了一个统一的编程模型来处理批处理、交互式查询、实时流处理和图计算等多种类型的数据处理任务。

2,spark特点

1)快速:基于内存的计算比 MapReduce 快 100 倍,基于磁盘快 10 倍。
2)易用:编写一个 spark 的应用程序可以使用 Java, Scala, Python, R,这就使得我们的开发非常地灵活。并且,对比MapReduce,spark 内置了 80 多个高级操作,这使得开发十分高效和简单。
3)运行范围广:spark 可以运行在 local、yarn、mesos、standalone、kubernetes 等多种平台之上。它可以访问诸如 HDFS, Cassandra, HBase, S3 等多种多样的数据源。
4)通用: spark 提供了 SparkSQL、 SparkStreaming、 GraphX、 MLlib 等一系列的分析工具。

3,spark功能

Spark Core:提供了Spark的基本功能,包括任务调度、内存管理、错误恢复等。
Spark SQL:提供了SQL接口和数据框架,用于处理结构化数据。
Spark Streaming:用于处理实时数据流,支持从Kafka、Flume等数据源接收数据。
MLlib:提供了大量的机器学习算法和工具,用于构建机器学习应用。
GraphX:用于图计算和图分析,提供了丰富的图算法和图操作。

4,spark优缺点

优点:

  1. 快速处理能力,基于内存
  2. 易于使用,支持多种语言
  3. 支持SQL查询,支持SQL和HQL
  4. 支持流式计算,Spark Streaming
  5. 可用性高,既支持Standalone独立部署(Master and Slaver模式),也支持基于Yarn的混合模式
  6. 丰富的数据源支持,Spark可以访问操作系统自身和HDFS,也可以访问Hive,HBase等

缺点:

  1. 不稳定,集群偶尔会挂掉。因为是基于内存运算的,因为确实与磁盘的I/O,如果数据量超出内存会出现挂掉的现象
  2. 数据的partition,会导致集群中的各台机器上计算任务分配不均匀
  3. 自身任务调度功能不够好

scala

一、Scala简介

Scala源自Java,构建在JVM之上,与Java兼容、互通。

Scala的优势:

1.多范式编程:

(1)面向对象编程:

  每个值都是对象;

  对象的数据类型和行为由类(Class)和特征(Trait,类似于interface)描述;

  利用特征实现混入式多重继承。

(2)函数式编程:

  每个函数都是一个值;

  支持高阶函数、柯里化(currying)、样例类(case class)及模式匹配......

2.Scala是静态语言,表达能力强,代码精简

3.Spark采用Scala语言设计,提供的API更加优雅,基于JVM的语言更融入Hadoop生态圈。

4.扩展性:隐式类、字符串差值

二、Scala变量(var)与常量(val)

变量:

    赋值后可以改变,生命周期中可以被多次赋值。

    var 变量名称:类型=xxx

    一般无需显示指定类型,Scala编译器会自动推断出类型

常量:

    赋值后不可变,类似于Java中的final变量

    val 常量名称:类型=xxx

三、Scala数据类型

Scala与Java有相同的数据类型,但是Scala的数据类型首字母全部大写

Byte、Short、Int、Long、Float、Double、Char、Boolean、String

1)Scala中一切数据都是对象,都是Any的子类;

2)Scala中数据类型分为两大类:数值类型(AnyVal)、引用类型(AnyRef),不管是值类型还是引用类型都是对象;

3)Scala数据类型仍然遵守,低精度的值类型向高精度值类型,自动转换(隐式转换);

4)Scala中的StringOps是对Java中的String增强;

5)Unit:对应Java中的void,用于方法返回值的位置,表示方法没有返回值。Unit是 一个数据类型,只有一个对象就是()。Void不是数据类型,只是一个关键字;

6)Null是一个类型,只有一个对象就是null。它是所有引用类型(AnyRef)的子类;

7)Nothing,是所有数据类型的子类,主要用在一个函数没有明确返回值时使用,因为这样可以把抛出的返回值,返回给任何的变量或者函数。

四,创建RDD

1.从内存读取数据创建RDD

parallelize()方法:通过调用SparkContext的parallelize方法,在一个已经存在的Scala集合上创建的(一个Seq对象)。集合的对象将会被拷贝,创建出一个可以被并行操作的分布式数据集。
例:

data = [1, 2, 3, 4, 5]

distData = sc.parallelize(data)

makeRDD()方法:只有scala版本的才有makeRDD,和parallelize类似。makeRDD方法实际上是将传入的集合和分区数两个参数传给parallelize方法然后将返回结果作为参数传给withScope方法调用
例:

val seq = Seq(1,Seq("iteblog.com","sparkhost1.com")),

(3,Seq("iteblog.com","sparkhost2.com")),

(2,Seq("iteblog.com","sparkhost3.com")),

val iteblog = sc.makeRDD(seq)

2.从外部存储系统读取数据创建RDD

通过HDFS文件创建RDD

例:

val test = sc.textFile("/opt/spark.txt")

二:对数据进行操作的方法

1.map()方法转换数据
2.sortBy()方法排序
3.collect()方法查询数据
4.flatMap()方法转换数据
5.take()查询某几个值
6.union()方法合并多个RDD
7.distinct()方法去重
8.intersection()方法用于求出两个RDD的共同元素
9.subtract()方法用于将前一个RDD中在后一个RDD出现的元素删除
10.使用键值对RDD的keys和values方法
11.使用键值对RDD的reduceByKey()方法
12.使用键值对RDD的groupByKey()方法
13.使用join()方法连接两个RDD
14.使用zip()方法组合两个RDD
15.使用combineByKey()方法合并相同键的值
16.使用lookup()方法查找指定键的值
17.读取和存储文件
读取:例
val input = sc.textFile("testjson.json")

查看内容:rdd.collect
存储:例
sc.parallelize(jsons).repartition(1).saveAsTextFile("json_out")

三:DataFrame基础操作

1.通过Parquet文件创建DataFrame

例:

val df = saprk.read.load("/opt/user.parquet")

2.通过json文件创建DataFrame

例:

val df = saprk.read.format("json").load("/opt/user.json")

3.通过外部数据库创建DataFrame

4.通过RDD创建DataFrame

方法一:

定义一个样例类

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

读取文件创建RDD

val data = sc.textFile("/opt/test.txt").map(_.split(","))

RDD转成DataFrame

val user = data.map(p => person(p(0),p(1),p(2).toInt)).toDF()

方法二:

五:DataFrame基础操作

1.通过Parquet文件创建DataFrame

例:

val df = saprk.read.load("/opt/user.parquet")

2.通过json文件创建DataFrame

例:

val df = saprk.read.format("json").load("/opt/user.json")

3.通过外部数据库创建DataFrame

4.通过RDD创建DataFrame

方法一:

定义一个样例类

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

读取文件创建RDD

val data = sc.textFile("/opt/test.txt").map(_.split(","))

RDD转成DataFrame

val user = data.map(p => person(p(0),p(1),p(2).toInt)).toDF()

方法二:

5.查看DataFrame数据

1.printSchema:输出数据模式
2.show():查看数据:默认查看20行
例:

user.show() //显示前20条数据

user.show(50) //显示前50条数据

3.first()/head()/take()/takeAslist():获取若干条记录
4.collect()/collectAslist():获取所有数据

6.DataFrame查询操作

1.where()方法:查询符合指定条件的数据
例:

val userwhere = user.where("age = 18 and gender = 'f' ")

2.filter()筛选符合条件的数据
例:

val userwhere = user.filter("age = 18 and gender = 'f' ")

3.select()/selectExpr()/col()/apply()方法
select:例:

val userSelect = user.select("userId","gender") //显示userId,gender字段

selectExpr:例:

val userSelect = user.selectExor("userId","gender as xx") //显示userId,gender字段,并把gender字段取名为xx

4.col()/apply()方法:获取指定字段,但只可以获取一个字段,返回的是一个Column对象
例:

val userSelect = user.col("userId") //显示userId字段

val userSelect = user.apply("userId") //显示userId字段

5.limit()方法:指定获取前n条记录
例:

val user = user.limit(3) //返回前3条记录

6.order()/sort()方法:根据指定字段进行排序,默认为升序
val user = user.orderBy(desc("ID")) //根据id降序排列

val user = user.orderBy(asc("ID")) //根据id升序排列

7.groupBy()方法:根据指定字段分组操作
例:

val user = user.groupBy(user("gender")).count //根据gender字段对user进行分组,并计算元素个数


本文转载自: https://blog.csdn.net/x11__111/article/details/138972773
版权归原作者 x11__111 所有, 如有侵权,请联系我们删除。

“spark方法”的评论:

还没有评论