spark的相关知识点

2024.6.19

1.scala：语言

2.spark：框架（jar包）

3.spark streaming：kafka

4.spark mlib：机器学习 算法

5.解释

1. Class：类
2. Case Class：样例类
3. Object：对象
4. User：类
5. New User（）：对象

6.scala设定变量

var name: String = “tom”:前面是名字，后面是类型

7.*def main(args: Array[String]): Unit = {
print("hello world"*)
}

Unit是返回值类型，可以写成int等类型

8.scala用法快捷键：1 to 100.for，是1到100的for循环的意思

9.scala的var能自己判断内容类型

10.var：可变和val：只读

11.Boolean的默认值是false，int的默认值是0，默认值的写法可以用’_’代替

12.如果class和object的名字是同一个名字的话，那么类就叫做伴生类，对象就叫做伴生对象。伴生对象相当于java里的static（静态方法），直接用类名方法就可以了。

13.def main(args: Array[String]): Unit = {
print(f"有一个学生叫$name,他的学号是$studentId,他$age 岁了,他有$height%.2f 米")
}

14.a=>b:可以理解成a代表入参，b代表返回值，无返回值的话”：Unit”可以去掉

2024.6.20

//集合不可变（当添加某一个符号后，会新建一个数组，以前的数组并不会改变）

var array=Array(1,2,3)
val str: String = array.mkString("~")
println(str)

mkString可以将数组装成字符串

Array(2)=33:可以根据下标修改值

Array.toList:可以将数据转换成List

2.可变数组

原理：在原来的数组层面上去更改

函数：ArrayBuffer、insert

例子：val buffer = ArrayBuffer(1,2,3)

//修改原来的数组

Buffer.update(2,33)

3.不可变数组

原理：原来的数组不变，再新增一个数组

4.函数的使用

函数:Insert（位置，数值）

例子：Buffer.insert(0,-1)

5.元组的各种情况

(1) 1元组

Tuple1（1）

(2) 2元组

Tuple（“tom”,18）

(3) 3元组

Tuple（“tom”,18,“天津”）

6.List(1,2,3,4,5)

1. 在左边追加元素

List.+:(0)

1. 在右边追加元素

List.:+(6)

(3)合并数组

例子：var a=List（1，2，3）

var b=List（4，5，6）

Var c=a ::: b

c就是合并后

如果编程::,那就是给b增加一个元素；

:::可以变成++，效果是一样的

6.函数

1. Filter
2. Sum
3. Map

7.ListBuffer是可变的List

不可变List如何变成可变List：list.toBuffer

8.Set

Set默认是不可变集合，，数值是不可重复的

特定：无序不可重复

例子：val set = Set(1,2,3,4,5,6)

删除元素：val ints = set-3(将元素3删除)

为什么不能用下标呢？

因为set集合是无序的

Var c1=a | b:求并集

Var c2=a & b：求交集

如果想要将set变成可变的集合

Val set1 =mutable.set

变成可变集合后

删除q元素的方法如下

1. Set1-=”q”
2. Set1.remove(“q”)

9.map集合++运算后，如果key相同，第二个会将第一个替换掉

10.从键盘读取数值

val input = StdIn.readInt()

例子如下：

object TestCore3 {
def main(args: Array[String]): Unit = {
val input = StdIn.readInt()
if (input>15){
print("你输入的值大于15")
}
}

2024.6.21

1.入参=>反参

2.mapValues:只对元素里的value做转换

3.flatMap:会把List的元素查出来，再合并，返回的类型是List，通俗的讲是将多个List合成一个List（扁平化）

4.scala里的固定方法

map

mapValues

flatMap

sortBy

filter

groupBy

distinct

sum

max

min

size

product

Reduce

5.构造器是new对象用的

构造器的入参是给属性赋值用的

有参构造器再创建对象的同时给属性赋值

2024.6.22

1.抽象体方法可以设置有参和无参，没必要设置反参，不然就没必要设置抽象方法了

2.抽象类里的方法叫做抽象方法，不需要些abstract

2024.6.25

1.名字,cache():将名字缓存起来，节省时间

2.val peopleDFCsv:DataFrame = spark.read.format("csv")

.option("sep",",")//文件分隔符，默认是逗号

.option("inferSchema","true")

.option("nullValue","???")//把???当作null处理

.option("header", "true")//第一行是不是有表头

.load("examples/src/main/resources/people.csv")//文件路径

Var peopleDFCsv2:DataFrame = peopleDFCsv.toDF(“字段名1”,“字段名2”)

上面的代码是为了给文件数据加表头

Val df3 = peopleDFCsv2.select(“字段名1”，“字段名2”)

dfs.show():只查询字段1和字段2的数据

如何过滤null值呢？

Val dfs4=dfs3.na.drop()

dfs4.show()

dfs4里的null值被过滤

3.dfs4.createOrReplaceTempView(“表名”):可以给文件设置表名

4.rdd3.saveAsTextFile("hdfs://node00:/month6/guojiahui/data1")

以上代码可以将rdd数据保存到hdfs中

2024.6.26

1.scala里的sql查询例子如下：

****import ****org.apache.spark.sql.SparkSession

*object *stu626sql {
*def *main(args: Array[String]): Unit = {
*val spark = SparkSession
.builder()
.appName("Spark SQL basic example"*)
.master(*"local"*)
.config(*"spark.some.config.option"*, "some-value")
.getOrCreate()
****import ****spark.implicits._
//读取数据

• **val peopleDFCsv = spark.read.format("csv"*)
  .option("sep", ",")
  .option("inferSchema", "true")
  .option("header", "false")
  .load("./input/city")
  //给数据设置表头
• **val df2 = peopleDFCsv.toDF("two_jz"*, "city1", "city2", "name", "bm")
  // val df3 = df2.select("city1", "city2")

** // df3.show()**

• df2.createOrReplaceTempView("city"*)
  //查询数据
• *spark.sql(
  """

**** |****
**** |select * from city****
**** |****
**** |****
**** |"""****.stripMargin).show()
}
}

2.reduceBykey(+)

上面的函数作用是先groupByKey，再对元素里的value（List）进行累加

2024.6.27

1.kafa作用原理

Mysql（增、删、改）经过日志解析的动作到kafka，通过kafka解耦合、缓解服务器压力到hdfs中，说白了，kafka的作用就是暂存数据

2.kafka：消息队列、消息中间件

3.kafka可以安装在多台电脑上

4.kafka可以把文件切分为多份，存到不同的电脑上，每个文件都有备份（备份到别的电脑上）

5.重点：

1. 如何启动和关闭zk、kafka
2. 命令创建topic
3. 编写生产者
4. 编写消费者

6.创建 topic

bin/kafka-topics.sh --bootstrap-server node00:9092 --create --topic 2203a --partitions 1 \

--replication-factor 1

查看topic 列表

bin/kafka-topics.sh --bootstrap-server node00:9092 --list

删除topic

bin/kafka-topics.sh --bootstrap-server node00:9092 --delete --topic 2203a

查看topic 详情

bin/kafka-topics.sh --bootstrap-server node00:9092 --topic 2203a --describe

2024.6.28

1.context.makeRDD(“变量”)

context.makeRDD("变量") 的作用是将给定的“变量”创建为一个弹性分布式数据集（RDD）。

在 Spark 中，RDD 是数据的基本抽象和处理单元。通过 context.makeRDD 方法，可以将各种数据来源（如本地数据集合、外部数据源等）转换为 RDD，以便在分布式环境中进行并行处理和计算。

例如，如果“变量”是一个数组 [1, 2, 3] ，那么通过 context.makeRDD 就将其转换为一个可以在 Spark 集群中并行处理的 RDD。

再比如，如果“变量”是一个字符串列表 ["apple", "banana", "cherry"] ，同样可以使用 context.makeRDD 将其转换为 RDD 来进行后续的诸如数据过滤、映射、聚合等操作。

2.zipWithIndex()：

zipWithIndex() 方法在 Spark 中的作用是为 RDD 中的每个元素分配一个从 0 开始的唯一索引，并将元素与其对应的索引组合成一个新的元组。

例如，对于一个包含元素 [10, 20, 30] 的 RDD ，使用 zipWithIndex() 后，得到的结果将是 [(10, 0), (20, 1), (30, 2)] 。

它在很多场景中非常有用，比如：

1. 当需要根据元素在原始 RDD 中的位置进行后续的处理或计算时。
2. 用于为数据添加唯一标识，以便在后续的操作中进行区分和跟踪。
3. 假设我们有一个包含学生成绩的 RDD [85, 90, 78] ，通过 zipWithIndex() 可以将成绩与其在 RDD 中的位置关联起来，方便进行进一步的分析，比如找出成绩最高的学生在原始数据中的位置。

2024.6.28

1.var a=new StreamingContext(conf,Second(10))

程序数据的微批处理，每隔10秒钟处理一次数据，积攒10秒钟的数据

StreamingContext 是 Spark 中用于创建和管理流式计算上下文的类。

conf 一般是一个配置对象，用于设置各种与流式处理相关的参数，比如应用名称、主节点地址、执行环境的配置等。

Second(10) 很可能是指定了批处理的间隔时间为 10 秒。这意味着 Spark 流式处理将会每隔 10 秒对收到的数据进行一次处理和计算。

例如，如果我们正在处理实时的传感器数据，每 10 秒处理一次新到达的数据，就可以使用这样的设置来控制处理的频率和节奏，以达到最佳的性能和准确性平衡。

在实际应用中，还需要根据数据量、处理逻辑的复杂度以及系统资源等因素来合理调整这个批处理间隔时间。

var b=a.socketTextStream(“node00”,9999)

程序要接受9999端口发过来的数据

在这段代码 var b = a.socketTextStream("node00", 9999) 中：

socketTextStream 通常是一个用于创建从指定主机（"node00"）和端口（9999）接收文本数据的流的方法。

这意味着程序正在尝试建立与名为 "node00" 的主机上的 9999 端口的连接，并以文本流的形式接收数据。

例如，如果这是一个实时监控系统，它可能会从该端口不断接收服务器发送的状态信息文本，并对其进行后续处理和分析。

又或者在一个分布式数据采集系统中，"node00" 上运行的某个服务会将采集到的数据通过 9999 端口以文本形式发送出来，这段代码所在的程序就负责接收和处理这些数据。

Spark Streaming 的好处主要体现在以下几个方面：

高吞吐量和低延迟

能够处理大规模的数据流，在保证较高数据处理速度的同时，还能将延迟控制在可接受的范围内。例如，在实时监控网络流量或处理金融交易数据时，能够快速响应并及时处理大量的数据。

容错性强

具备良好的容错机制，当节点出现故障时能够自动恢复数据处理，确保数据处理的连续性和准确性。例如，在分布式计算环境中，某个计算节点突然宕机，Spark Streaming 可以从其他节点恢复丢失的数据和计算进度。

与 Spark 生态系统集成紧密

可以方便地与其他 Spark 组件（如 Spark SQL、Spark MLlib 等）进行交互和集成，构建复杂的数据处理流水线。比如，先通过 Spark Streaming 接收实时数据，进行初步处理后，再使用 Spark SQL 进行复杂的查询和分析，或者使用 Spark MLlib 进行实时的机器学习模型训练。

易于开发和维护

提供了简洁直观的编程接口，开发人员可以使用熟悉的编程语言（如 Java、Scala 和 Python）进行开发。而且，代码的可读性和可维护性较高。

支持多种数据源和数据格式

能够从各种各样的数据源（如 Kafka、Flume、Socket 等）获取数据，并支持多种数据格式（如 JSON、CSV 等）的处理。例如，从 Kafka 主题中读取实时消息，并将其解析为特定的格式进行后续处理。

可扩展性好

可以轻松地在集群中扩展计算资源，以应对不断增长的数据量和计算需求。通过增加节点数量，可以线性地提高系统的处理能力。

2024.7.1

1.spark core(spark框架 提供的API，类和方法)

rdd复杂 灵活度高

2.rdd.saveAsTextFile(“路径”)：将rdd数据保存到路径位置

3.jdbcDF.write

.format("jdbc")

.option("url", "jdbc:postgresql:dbserver")

.option("dbtable", "schema.tablename")

.option("user", "username")

.option("password", "password")

.mode(SaveMode.Overwrite)//保存模式 覆盖

.save()

mode(SaveMode.Overwrite)//保存模式：覆盖

如果不写覆盖的保存模式，之后又可能会报错

4.合理设置日志打印级别,正常输出为“info”，错误为“ERROR”

spark.sparkContext.setLogLevel(“error”)

5.开启checkpoint机制（缓存RDD保存到HDFS中）

Spark.sparkContext.setCheckpointDir(“hdfs路径”)

6.spark streaming

streaming:数据的流式处理

流式处理和批量处理的区别？

流式处理：有源源不断的数据 进入做分析

批量处理：现有一批数据 做分析

有界数据和无界数据？

有界数据：是指具有明确开始和结束点的数据集合

无界数据：没有明确的开始或结束，是持续不断生成和流动的数据

实时处理和离线处理？

2024.7.2

1.sparkstreaming：spark流

处理源源不断的数据

它依赖于kafka

2.kafka里的topic

分区：对文件分块存储

副本：对每个分区的数据进行备份

//设置背压机制：调整消费者消费的速率

conf.set("spark.streaming.backpressure.enabled", "true")

//每个分区消费消息的个数的速率

conf.set("spark.streaming.kafka.maxRatePerPartition", "2000")  

//优雅关闭

conf.set("spark.streaming.stopGracefullyOnShutdown", "true")

//设置sparkstreaming消费批次间隔为5s

val ssc = new StreamingContext(conf, Seconds(5))

4.2024.7.3

1.intersection:去交集

数据1.intersection（数据2）获取返回值，可以取数据1和数据2的交集

2.producer.send(new ProducerRecord<String, String>("xuang", str));

send发送只能时一行一行的发送

3.sparkmilb（机器学习）

什么是机器学习呢？

机器学习是人工智能的一个分支，它是从大量的数据中发现规律，提取知识，并在实践中不断地完善和增强自我。

4.机器学习模型=数据+机器学习算法

5.线性回归

6.求导：

求导是数学中微积分的一个重要概念。

导数（也称为导函数）表示一个函数在某一点处的变化率。

具体来说，给定一个函数 (f(x))，它的导数 (f'(x)) 描述了函数值 (f(x)) 随着自变量 (x) 的变化而变化的快慢程度。

例如，对于函数 (f(x) = x^2)，其导数 (f'(x) = 2x) 。这意味着在 (x) 处，函数值的变化率是 (2x) 。当 (x = 1) 时，变化率为 (2) ；当 (x = 2) 时，变化率为 (4) 。

求导在许多领域都有广泛的应用，比如物理学中，速度是位移的导数，加速度是速度的导数；在经济学中，边际成本、边际收益等概念都与导数相关；在工程学中，求导可用于优化设计、分析系统的动态特性等。

求导的方法包括基本函数的求导公式（如幂函数、指数函数、对数函数等）、四则运算的求导法则（如加法法则、乘法法则、除法法则）以及复合函数的求导法则（链式法则）等。

2024.7.4

机器学习：用已知的数据推断规律

2024.7.5

1.分类评估器：Y值是有限的

2.回归评估器：Y值是无限的

2024.7.6

checkpoint：把数据持久化到hdfs

2024.7.8

1.算法有三种

逻辑算法

线性算法

决策树算法

2.val trainingSummary = lrModel.summary

如果您使用的是线性回归模型或逻辑回归模型，val trainingSummary = lrModel.summary 这行代码获取的是线性回归模型训练或逻辑回归模型训练的摘要信息。

这个摘要信息可能包含一些关于模型训练过程和结果的统计量和评估指标，例如均方误差（MSE）、均绝对误差（MAE）、决定系数（R-squared）等，具体取决于所使用的线性回归库和实现。这些指标可以帮助您评估模型的拟合效果和预测能力。

3.回归和分类问题

回归：线性回归和决策树

分类：逻辑回归和决策树

评估器：回归评估器（mae、mse、r2、误差）

分类评估器（准确率 召回率）

交叉验证：代码多、复杂

2024.7.10

1.dataframe格式可以装成dataset格式再转成rdd格式

例子：val order_info_rdd = order_info.as[Order_info].rdd

Order_info是我们提前创建好的对象

order_info.as[Order_info]：将dataframe格式转成dataset格式

order_info.as[Order_info].rdd:将dataframe格式转成rdd格式

2.如果将数据写入navicate中时，我们有警告，可以再数据库名后写上“?useSSL=false”

3.dataframe可以转为dataSet再转为rdd格式

写法如下（OrderInfo事先定义的对象）

val ds2 = df2.as[OrderInfo].rdd

4.DataFrame：

DataFrame 是一种以命名列的方式组织的分布式数据集。它类似于关系型数据库中的表，可以通过指定列名和数据类型来定义。

特点：

提供了丰富的结构化操作，如选择、过滤、聚合等。

可以与 Spark SQL 进行无缝集成，方便使用 SQL 语句进行数据处理。

具有优化的执行引擎，能够自动进行一些性能优化。

例如，如果您有一个包含用户信息的 DataFrame，您可以轻松地使用 df.filter($"age" > 20) 来筛选出年龄大于 20 岁的用户。

DataSet：

DataSet 是 DataFrame 的扩展，它在 DataFrame 的基础上，提供了类型安全和面向对象的编程接口。

特点：

提供了强类型检查，减少了运行时错误的可能性。

对于特定类型的数据处理，性能可能更优。

例如，如果您定义了一个 case class User(name: String, age: Int) ，然后可以创建一个 DataSet[User] 来处理用户数据。

RDD（Resilient Distributed Dataset）：

RDD 是 Spark 最基础的数据抽象，表示一个不可变、可分区、里面的元素可并行计算的集合。

特点：

提供了丰富的底层操作接口，如 map、filter、reduce 等。

具有很强的容错性和弹性。

例如，使用 rdd.map(_ * 2) 可以将 RDD 中的每个元素乘以 2 。

不同之处：

类型安全：DataFrame 和 DataSet 具有一定的类型安全特性，而 RDD 是无类型的。

性能优化：DataFrame 和 DataSet 能够利用 Catalyst 优化器进行更多的性能优化。

编程接口：RDD 提供的是更底层的操作接口，而 DataFrame 和 DataSet 提供了更高级和结构化的操作方式。

与 Spark SQL 的集成：DataFrame 与 Spark SQL 的集成更加紧密，更方便使用 SQL 进行数据处理。

在实际应用中，选择使用哪种数据结构取决于具体的需求和场景。如果需要类型安全和更高级的结构化操作，通常会选择 DataFrame 或 DataSet；如果需要更底层的控制和自定义操作逻辑，可能会选择 RDD 。

5.kafka API将读取到的数据从文件发送到kafka

这里是再sparkstreaming消费者里使用，将文件里的数据读出来，再发送到kafka的topic中。具体操作如下：

1>FileReader fileReader = new FileReader(“文件的绝对路径位置”)

2>List<String> string = fileReader.readLines();

遍历出的内容就是文件的内容了

6.将json文件的内容装成一个对象

ValDS1=Stream.map(Record=>jsoN.parse0bject(record.value()，classOf[Device]))

Record=>jsoN.parse0bject(record.value()，classOf[Device])是要学习的

解释如下：

这段代码是使用 Java 中的JSON库来将一个JSON格式的字符串record.value()解析为一个Device类的对象。

JSON.parseObject是JSON库中的一个方法，用于将JSON字符串转换为指定类型的对象。

在这段代码中：

record.value() 应该是获取到的一个表示JSON数据的字符串。

classOf[Device] 是指定了要将解析后的JSON数据转换为Device类的对象。这要求Device类具有与JSON字符串中的字段相对应的属性和构造函数，以便能够正确地进行映射和赋值。

例如，如果Device类有属性id、name和status，并且JSON字符串record.value()的值类似于 {"id": 1, "name": "Device1", "status": "active"} ，那么通过JSON.parseObject(record.value(), classOf[Device]) 就可以将这个JSON字符串转换为一个Device对象，其中对象的id属性值为 1，name属性值为 "Device1"，status属性值为 "active" 。

7.如果想要将许多条数据转换成rdd形式并遍历

文件.foreachRDD(rdd1=>rdd1.saveAsTextFile(“hdfs路径”))

代码是将遍历的内容保存到hdfs当中

8.df内容如何转为json类型并保存到路径位置

表名.write.format(“json”).save(“路径位置”)

9.scala里如何创建udf自定义函数将日期转换为yyyy-MM-dd的格式

Spark.udf.register(“别名”,(字段名:类型名)=>{

DateUtil.format(字段名,”yyyy-MM-dd”)

})

查询语句使用别名来检查自定义函数是否成功！

8.valsession=DbUtil.newSession(newSimpleDateSource(“jdbc:mysql://localhost:3306/数据库名”,”root”,”123456”))

Session.execute(“insert into demo set a=?”,o._1._1)

9.如果相同时定义一个core和sql，那么要注意sql要写在core的下方，不然可能会报错

10.中文提词器

个人认为是个难点且重点，用法如下

(str(0), str(1), ToAnalysis.parse(str(2).replace("[", "").replace("]", "")).toStringWithOutNature, str(3), str(4), str(5))

ToAnalysis.parse(str(2).replace("[", "").replace("]", "")).toStringWithOutNature

ToAnalysis是一个工具类，解析str（2）内容，这里的str（2）把多余的部分去掉了，通过replace替换来去除“[”和“]”,这些工作做完后，解析出来的内容其实有\t、\m这些多余的东西，我们也要将它们去除掉，就用到了toStringWithOutNature来实现

11.sparksql的拆分数据（不建议记住，因为许多都要背）

select *

from movie

lateral VIEW

explode(split(type,"/")) movie_info_tmp AS type_name ) t

这段 SQL 语句的作用是从 movie 表中查询所有列，并使用 LATERAL VIEW explode(split(type, "/")) 来对 type 列进行拆分和展开。

具体解释如下：

SELECT * FROM movie：选择 movie 表中的所有列。

LATERAL VIEW explode(split(type, "/")) movie_info_tmp AS type_name：使用 explode 函数对 split(type, "/") 的结果进行展开。split(type, "/") 会将 type 列的值按照 / 进行分割，形成一个数组。explode 函数会将这个数组中的每个元素展开成一行，结果存储在临时表 movie_info_tmp 中，其中展开后的列名为 type_name。

最终，查询结果将包含 movie 表的所有列以及展开后的 type_name 列。

2024.7.12

1.map：替换旧的元素值

2.flatmap：返回一个新的RDD

例如：list换成list，元素扁平化，变成新的RDD

把方程的返回值拆开放到全新的List中