spark学习笔记

idea maven加速

右键项目中的pom.xml等，在Maven中选择修改setting.xml，如果没有这个配置文件会有create setting.xml的选项，

<?xml version="1.0" encoding="UTF-8"?>
<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">
    <mirrors>
        <mirror>
            <id>alimaven</id>
            <name>aliyun maven</name>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
            <mirrorOf>central</mirrorOf>
        </mirror>

        <mirror>
            <id>uk</id>
            <mirrorOf>central</mirrorOf>
            <name>Human Readable Name for this Mirror.</name>
            <url>http://uk.maven.org/maven2/</url>
        </mirror>

        <mirror>
            <id>CN</id>
            <name>OSChina Central</name>
            <url>http://maven.oschina.net/content/groups/public/</url>
            <mirrorOf>central</mirrorOf>
        </mirror>

        <mirror>
            <id>nexus</id>
            <name>internal nexus repository</name>
            <url>http://repo.maven.apache.org/maven2</url>
            <mirrorOf>central</mirrorOf>
        </mirror>

    </mirrors>

</settings>

添加如上的mirrors

Spark 依赖

这部分需要在项目的pom.xml中进行配置，主要是增加了dependencies那部分内容:

<?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">
    <parent>
        <artifactId>testclasses</artifactId>
        <groupId>com.test</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>spark-core</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.1.2</version>
        </dependency>
    </dependencies>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

</project>

要注意的是，可以通过右击pom.xml然后maven然后reload project来完成这部分依赖的下载

HADOOP 环境依赖

如果没有HADOOP环境依赖就会报一个恶心的错误 Could not locate executablenull\bin\winutils.exe in the Hadoop binaries以及这个诱发的奇奇怪怪的错误。可能可以采用https://github.com/steveloughran/winutils；实操使用的是尚硅谷给的配置文件，已经上传到了百度网盘。

1 2	链接：https://pan.baidu.com/s/1z23GQYaPmzq7VDS_f2646A 提取码：piza

然后配置一下环境变量，HADOOP_HOME 给 hadoop-3.1.0；然后Path环境变量加一个%HADOOP_HOME%\bin，重启IDEA就行了

wordcount

这部分参考了尚硅谷的教程，文件结构如下图：

代码如下:

package com.test.spark.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object spark01_wordcount {
  def main(args: Array[String]): Unit = {
    //Application
    //Spark框架
    //TODO 建立和Spark框架的连接
    //JDBC：Connection
    // 创建 Spark 运行配置对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    // 创建 Spark 上下文环境对象（连接对象）
    val sc = new SparkContext(sparkConf)

    //TODO 执行业务操作
    //1.读取文件，获取一行一行的数据
    //Hello World
    val lines: RDD[String] = sc.textFile("datas")

    //2.将一行数据进行拆分，形成一个个的单词（分词=>扁平化：将整体拆分成个体的操作）
    //"Hello World" => Hello,World...
    //val words: RDD[String] = lines.flatMap(s => s.split(" "))
    val words: RDD[String] = lines.flatMap(_.split(" "))

    //3.将数据根据单词进行分组，便于统计
    //(hello,hello...),(world,world,world...)
    val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word)
    //4.将分组后的数据进行转换
    //(hello,hello...),(world,world,world...)
    //(hello,2),(world,3)
    val res: RDD[(String, Int)] = wordGroup.map {
      case (word, list) => {
        (word, list.size)
      }
    }

    //5.将转换的结果采集到控制台打印出来
    val array: Array[(String, Int)] = res.collect()
    array.foreach(println)

    //TODO　关闭 Spark 连接
    sc.stop()
  }
}

scala快速入门

因为spark由scala编写而成，因此快速入门一下scala,学习了这个专栏https://zhuanlan.zhihu.com/p/32859134

scala是一种函数式语言，其函数也能当成值来使用。Scala提供了轻量级的语法用以定义匿名函数，支持高阶函数，允许嵌套多层函数，并支持柯里化。Scala的case class及其内置的模式匹配相当于函数式编程语言中常用的代数类型。可以利用Scala的模式匹配，编写类似正则表达式的代码处理数据。

从并发角度而言：Scala使用Actor作为其并发模型，Actor是类似线程的实体，通过“邮箱”发收消息。Actor可以复用线程，因此可以在程序中可以使用数百万个Actor，而线程只能创建数千个。在2.10之后的版本中，使用Akka作为其默认Actor实现。

主要内容

变量
函数
数组
映射和元组转换
类与对象
继承与特质
集合
模式匹配和样例

变量

不强制指定变量的类型可以编译器推断出来
1
2
scala> 8*5
res0:Int=40
val值不能改变它的内容，鼓励使用val
可变的变量用var
变量或者函数的类型写在变量或者函数的后面，有点像go
1
val test:String="test"
不需要使用分号最后，仅当同一行代码中存在多条语句时才需要分号隔开。
常用的数据类型与Java一样，Byte、Char、Short、Int、Long、Float、Double及Boolean，这些都是类。
+-*/%等操作符实际上是方法。
对于BigInt和BigDecimal对象，可以以常规的方式使用数学操作符（但在Java中同样的操作要写成x.multiply(x)）。
1
2
scala>val x:BigInt=11111111222222222333333333333
scala>x*x

函数

表达式有值，语句执行动作。
函数式中不使用return。
语句中，分号不是必需的。
代码块也有值，最后一个表达式就是值。if表示式有值。
Scala中，几乎所有构造出来的语法结构都有值，不像Java中把表达式和语句（if语句）分为两类。

条件表达式

在Scala中if/else表达式有值，这个值就是在if或else之后的表达式的值。

scala > var x = 10 
x : Int = 10 
scala > val r = if ( x > 0 ) 1 else - 1 
r : Int = 1 
scala > var x = 0 
x : Int = 0 
scala > val r = if ( x > 0 ) 1 else - 1 
r : Int = - 1

可能if没有输出值，但在Scala中，每个表达式都有某种值。

scala > var x = 0 
x : Int = 0 
scala > val r = if ( x > 0 ) 1 
r : AnyVal = ()## 因为x=0，所以此时未定义

块表达式

在Scala中{}块包含一系列表达式，其结果也是一个表达式。块中最后一个表达式的值就是块的值。
对于某个val的初始化需要分多步完成的情况很实用。

1	val dis = { val dx = x - x0 ; val dy = y - y0 ; sqrt ( dx * dx + dy * dy )}

循环

while 循环和java一样
for循环：

1
2
3

for ( i <- 1 to n ) { 
	r = r * i 
}

1 to n：[1,n]
1 until n:[1,n)

增强for循环

可以多个生成器，用分号隔开

1	scala > for ( i <- 1 to 3 ; j <- 1 to 3 ) print (( 10 * i + j ) + " " )

每个生成器都可以带一个守卫，以if开头的Boolean表达式 (if前并没有分号)

1	scala > for ( i <- 1 to 3 ; j <- 1 to 3 if i != j ) print (( 10 * i + j ) + " " )

for推导式：for循环的循环以yield开始，则该循环会构造出一个集合，每次迭代生成集合中的一个值。和python yield一样
1
2
scala> for(i < -1 to 10) yield i%3
res2:scala.collection.immutable.IndexedSeq[Int]=Vector(1,2,0,1,2,0,1,2,0,1)

函数

需要给出函数名、参数和函数体
必须给出所有参数的类型
1
def abs(x:Double)= if (x>=0)x else -x

递归函数必须指定返回值类型

1	def fac(n:Int):Int = if(n<=0)1 else n* fac(n-1)

不需要return
用=连接函数体

默认参数以及带名参数

1
2
3

scala>def decorate(str:String,left:String="[",right:String="]")=left+str+right
scala>decorate("test")
scala>decorate("test","{","}")

也可以指定参数，此时顺序可以不一样
可以混用未命名以及指定参数，未命名的排在前面，和C++一样

可以实现接受可变长列表的参数，用*

scala>def sum(arg:Int*)={
	var result=0
	for(arg <- args)result+=arg
	result
	}
scala>val s = sum(1,2,3,4)
s: Int = 10

可以用_*把一个整数区间转换成参数序列：

对于上面的sum，应该输入一个列表，因此输入sum(1 to 5)是错的，需要输入sum(1 to 5: _*)
如果函数体包含在花括号当中，但没有前面的 =号，返回类型是Unit，这样的函数被称做过程。过程不返回值，调用它仅仅是为了它的副作用。

当val被声明为lazy时，它的初始化将被推迟，直到首次对它取值。

1
2

lazy val words = scala . io . Source . fromFile ( "/usr/share/dict/words" ). mkString
##可以故意把文件名写错，试一下在初始化语句被执行的时候会不会报错（只有访问words时才提示文件未找到）

数组

长度固定使用Array，长度有变化使用ArrayBuffer
提供初始值时不要使用new
用()来访问元素
for(elem <- arr)遍历元素
for(elem <- arr if …) yield …将原数组转为新数组

定长数组

整数初始化为0，string用new初始化为null，提供初始值不需要new，用()访问（相当于C++[]）
1
2
3
scala> val nums = new Array[Int](10)
scala> val str1 = Array("Hello", "Scala")
scala> val s = str1(0)

变长数组

使用ArrayBuffer，需要import scala.collection.mutable.ArrayBuffer
初始化:
1
scala> val b = ArrayBuffer[Int]()
+=追加元素或者多个用括号包起来的
1
2
scala> b+=1
scala> b+=(2,3)
++=追加任何集合
1
scala> b++=Array(4,5,6)
用b.trimEnd(2)移除最后两个元素

可以在任何地方插入或者移除(不高效，需要移动后面的所有元素)

1	scala> b.insert(2,3,4) ## 在下标2的地方插入3,4，比如(1,2,9)变成了(1,2,3,4,9)

使用toArray变成定长，toBuffer变成变长

遍历

scala> for (i <- 0 until b.length)
		println(i + ":" + b(i))
##或者不用下标
scala>  for(elem <- b)
		println(elem)

数组转换

for推导式，从一个数组转换，生成一个全新的数组

1 2	scala> val a = Array(2, 3, 5, 7) scala> val res = for(elem <- a) yield 2 * elem

ArrayBuffer同理

映射和元组转换

映射

这是键值对的集合，相当于字典

例子:

## 不可变映射，值不能够被改变
scala> val scores = Map("Alice" -> 90, "Bob" -> 88)
## 可变映射
scala> val scores1 = scala.collection.mutable.Map("Alice" -> 90, "Bob" -> 88)
## 构造一个空映射，需要选定一个映射实现并给出类型参数(注意最后是 方括号[] )
scala> val scores2 = new scala.collection.mutable.HashMap[String, Int]
## 使用()获取键值
scala> val aliceScore = scores("Alice")
## 判断映射中是否包括某个指定键的值，用contains方法
scala> val bobScore = if(scores.contains("Bob")) scores("Bob") else 0
## 等价于:
scala> val bobScore1 = scores.getOrElse("Bob", 0)
## 可变映射中更新某个映射中的值或添加一个新的映射关系
scala> val scores1 = scala.collection.mutable.Map("Alice" -> 90, "Bob" -> 88)
scores1("Bob") = 99
scores1("Fred") = 79 ##Fred不存在，会添加一个新的关系
## 使用+=操作来添加多个关系(key不存在就添加，存在就更新)
scores1 += ("Yezhiwei" -> 100, "Fred" -> 90)
## 使用-=操作来移除某个键值对
scores1 -= "Bob"
## 不可变映射同样的操作不会改变原来的值，但是会返回一个新的对象
## 可以使用scores.keys或者scores.values来访问键或者值

元组

元组是不同类型的值的聚集，用圆括号包裹
1
scala> (1, 3.14, "Fred")

元组的组元从1开始，比如可以:

scala> (1, 3.14, "Fred")
res11: (Int, Double, String) = (1,3.14,Fred)
scala> res11._1
res12: Int = 1s

可以使用模式匹配来获取组元

1
2
3

scala> val (first, second, third) = res11
## 不需要的值用_
scala> val (first, second, _) = res11

元组可以用于函数需要返回不止一个值的情况

1 2	scala> "Hello Scala".partition(_.isUpper) res13: (String, String) = (HS,ello cala)

类定义与构造器

每一个类有一个主构造器，它执行类中所有的语句
类中自带getter以及setter方法(jvm getter/setter方法)
用@BeanProperty注解生成JavaBean的getXxx/setXxx方法
辅助构造器是可选的，它们都叫做this

类定义

例子：

scala> class Counter {
    | private var value = 0
    | def increment() {value += 1}
    | def current() = value
}

Scala生成面向JVM的类时，其中有一个私有的age字段及相应的两个公有的getter和setter方法；

对于私有字段，getter和setter方法也是私有的；

在Scala中，getter和setter分别叫做age和age_= (注意下划线和等号间没有空格)

重新定义getter和setter方法，如：年龄不能设置成比现在的更小

scala> class Person{
    | private var privateAge = 0
    | def age = privateAge
    | def age_= (newValue: Int) {
    | if (newValue > privateAge) privateAge = newValue
    | }

如何控制Scala对每个字段生成getter和setter的方法？

如果字段是私有的，则getter和setter方法也是私有的

如果字段是val，则只生成getter方法

如果不需要任何getter和setter，将字段声明为private[this]即可

用@BeanProperty注解生成JavaBean的getXxx/setXxx方法

1
2
3

scala> import scala.reflect.BeanProperty
scala> class Person{
    | @BeanProperty var name: String = _

将会生成四个方法：

name: String

name_=(new Value: String): Unit

getName(): String

setName(newValue: String): Unit

构造器

主构造器例子:

// Scalaclass 
Person(val name: String, val age: Int) {}
// 以上代码相当于Java的以下代码public 
class Person {    
	private String name;    
	private int age;    
	public Person(String name, int age) {        
	this.name = name;        
	this.age = age;    }    
	// getter and setter    
	...}

除了主构造器之外可以定义任意多的辅助构造器，辅助构造器的名称为this，每一个辅助构造器都需要以一个对先前定义的其他辅助构造器或者主构造器的调用开始

class Person {
	private var name = ""
	private var age = 0
	def this(name: String) {
		// 调用主构造器
		this()
		this.name = name
		}
		
	def this(name:String,age: Int){
		// 调用前一个辅助构造器
		this(name)
		this.age=age
		}
		
	}

我感觉是一种聚合的形式

高阶函数

同时可以参考这个博客

匿名函数,并且可以放到变量中,或者直接传递给另一个函数

1 2	scala > val test=(x:Double) => 3 * x scala > Array(3.14,1.42).map((x:Double)=>3*x)

在变量中存储非你梦函数需要_，这个表示这个函数:

1 2	scala > def fun ( msg : String ) { println ( msg )} scala > val f = fun _

带函数参数的函数：

scala >   def  mulBy ( factor :   Double )   =   ( x :   Double )   =>  factor  *  x
mulBy :   ( factor :   Double ) Double   =>   Double   

scala >  val q  =  mulBy ( 5 )  
q :   Double   =>   Double   =   <function1>   

scala >  q ( 20 )  res14 :   Double   =   100.0   
// val q = mulBy(5) 相当于 val q = (x: Double) => 5 * x
//

//函数可以作为参数，进行传递（大多数情况都是通过匿名函数的形式）
//定义一个函数calculator，该函数接收一个函数类型的参数op，op定义了对两个整数要进行什么样的操作
def calculator(a:Int,b:Int,op:(Int,Int)=>Int): Int ={
  op(a,b)
}

//定义一个函数f1,完成两个整数的加法运算
def op(a:Int,b:Int):Int={
  a + b
}
//println(calculator(10,20,op))
println(calculator(50, 20, (a: Int, b: Int) => {
  a - b
}))

println(calculator(50, 20, _ - _))
println(calculator(50, 20, _ + _))
*/

函数可以作为返回值进行返回:

函数的嵌套
函数链式调用，通过参数传递数据，在执行的过程中，函数始终占据栈内存，容易导致内存溢出

//函数可以作为返回值进行返回----函数的嵌套
    def f1():()=>Unit ={
      def f2(): Unit ={
        println("f2函数被调用了")
      }
      //将f2作为返回值返回给f1
      f2 _
    }

    //ff就是f2
    //var ff = f1()
    //ff()

    //f1()()

闭包

闭包是一个函数，返回值依赖于声明在函数外部的一个或多个变量。
闭包通常来讲可以简单的认为是可以访问一个函数里面局部变量的另外一个函数。
内层函数访问外层函数的局部变量，会自动延长外层函数局部变量的生命周期，与内层函数形成一个闭合的效果，我们称之为闭包

/*
    def f1(): (Int)=>Int ={
      var a:Int = 10
      def f2(b:Int): Int ={
        a + b
      }
      f2 _
    }
    */

    //执行f1函数返回f2,将返回的f2赋值给ff变量
    //val ff: Int => Int = f1()

    //闭包：内存函数f2要访问外层函数f1局部变量a，当外层函数f1执行结束之后，f1会释放栈内存，但是会自动的延长f1函数的局部变量的生命周期，
    //    和内层函数f2形成一个闭合的效果，我们将这种闭合的效果称之为闭包

    //如果存在闭包，那么编译器会生成包含$anonfun$的字节码文件

    //闭包  = 外层函数的局部变量  + 内层函数

    //调用ff函数，其实就是调用f2
    //println(ff(20))

    //也可以直接通过如下方式调用
    //println(f1()(30))


    println("-----------------------------------")
    //以上代码可以转换为如下代码
    /*
    def f3() ={
      var a:Int = 10
      (b:Int) => a + b
    }

    f3()(30)
    */

柯里化

将一个参数列表中的多个参数，拆分为多个参数列表
好处1：每一个参数列表表示函数清晰明确
好处2：简化闭包代码的编写

//柯里化
    //将一个函数的一个参数列表中的多个参数，拆分为多个参数列表
    //简化闭包代码的编写
    def f4()(b:Int):Int = {
      var a:Int = 10
      a + b
    }
    println(f4()(20))

    //f4在执行的时候，其实会转换为以下结构
    /*
    def f4() ={
      var a:Int = 10
      (b:Int) => a + b
    }

PS: Python 闭包中自由变量与全局变量的区别;python闭包

模式匹配和样例类

mathch表达式是一个更好的switch，不会有穿透到下一个分支的问题

与if类似，match也是表达式，而不是语句

object Test {
   def main(args: Array[String]) {
      println(matchTest(3))

   }
   def matchTest(x: Int): String = x match {
      case 1 => "one"
      case 2 => "two"
      case _ => "many"
   }
}

结果为:

1
2
3

$ scalac Test.scala 
$ scala Test
many

scala模式匹配不会自动进入下一个分支，不需要再每一个分支后面使用break
case _等价于default
match可以使用任意类型，不一定需要数字

还可以给模式添加守卫，比如匹配所有的数字:

object Test {
   def main(args: Array[String]) {
      println(matchTest(3))

   }
   def matchTest(x: Int): String = x match {
      case 1 => "one"
      case 2 => "two"
      case _ if Character.isDigit(x)=>Character.digit(x,10)
      case _ => "many"
   }
}

还可以对类型匹配，比如case x:Int=>x这样
另外还有一些技巧见原文

样例类，常用于模式匹配:

object Test {
   def main(args: Array[String]) {
        val alice = new Person("Alice", 25)
        val bob = new Person("Bob", 32)
        val charlie = new Person("Charlie", 32)
   
    for (person <- List(alice, bob, charlie)) {
        person match {
            case Person("Alice", 25) => println("Hi Alice!")
            case Person("Bob", 32) => println("Hi Bob!")
            case Person(name, age) =>
               println("Age: " + age + " year, name: " + name + "?")
         }
      }
   }
   // 样例类
   case class Person(name: String, age: Int)
}

输出：

$ scalac Test.scala 
$ scala Test
Hi Alice!
Hi Bob!
Age: 32 year, name: Charlie?

PS:在声明样例类时，下面的过程自动发生了：

构造器的每个参数都成为val，除非显式被声明为var，但是并不推荐这么做；
在伴生对象中提供了apply方法，所以可以不使用new关键字就可构建对象；
提供unapply方法使模式匹配可以工作；
生成toString、equals、hashCode和copy方法，除非显示给出这些方法的定义。

spark学习笔记