Spark基础内容

发布时间:2024年01月07日

Spark基本介绍

Spark是什么

定义

Apache Spark是用于大规模数据(large-scala data)处理的统一(unified)分析引擎.

Spark与MapReduce对比

mapreduce架构图如下:
在这里插入图片描述
MapReduce的主要缺点:

  • 1- MapReduce是基于进程进行数据处理,进程相对线程来说,在创建和销毁的过程比较消耗资源,并且速度比较慢
  • 2- MapReduce运行的时候,中间有大量的磁盘IO过程。也就是磁盘数据到内存,内存到磁盘反复的读写过程
  • 3- MapReduce只提供了非常低级(底层)的编程API,如果想要开发比较复杂的程序,那么就需要编写大量的代码。

Spark相对MapReduce的优点:

  • 1- Spark底层是基于线程来执行任务
  • 2- 引入了新的数据结构——RDD(弹性分布式数据集),能够让Spark程序主要基于内存进行运行。内存的读写数据相对磁盘来说,要块很多
  • 3- Spark提供了更加丰富的编程API,能够非常轻松的实现功能开发

线程和进程对比

在这里插入图片描述

  • 线程的创建和销毁,比进程会更加的快速,以及更加的节省资源
  • 进程很难共享内存中的数据;而同个进程中的线程可以共享内存中的数据

Spark特点

  • 高效性

    • 计算速度快
      • 提供了一个全新的数据结构RDD(弹性分布式数据集)。整个计算操作,基于内存计算。当内存不足的时候,可以放置到磁盘上。整个流程是基于DAG(有向无环图)执行方案。
      • Task线程完成计算任务执行
  • 易用性

    • 支持多种语言开发 (Python,SQL,Java,Scala,R),降低了学习难度
  • 通用性

    • 在 Spark 的基础上,Spark 还提供了包括Spark SQL、Spark Streaming、MLlib 及GraphX在内的多个工具库,我们可以在一个应用中无缝地使用这些工具库。
  • 兼容性(任何地方运行)

    • 支持三方工具接入

      • 存储工具
        • hdfs
        • kafka
        • hbase
      • 资源调度
        • yarn
        • Kubernetes(K8s容器)
        • standalone(spark自带的)
      • 高可用
        • zookeeper
    • 支持多种操作系统

      • Linux
      • windows
      • Mac

Spark框架模块

在这里插入图片描述

  • Spark Core API:实现了 Spark 的基本功能。包含RDD、任务调度、内存管理、错误恢复、与存储系统交互等模块。数据结构RDD。
  • Spark SQL:我们可以使用 SQL处理结构化数据。数据结构:Dataset/DataFrame = RDD + Schema。
  • Structured Streaming:基于Spark SQL进行流式/实时的处理组件,主要处理结构化数据。
  • Streaming(Spark Streaming):提供的对实时数据进行流式计算的组件,底层依然是离线计算,只不过时间粒度很小,攒批。–了解
  • MLlib:提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等。
  • GraphX:Spark中用于图计算的API,性能良好,拥有丰富的功能和运算符,能在海量数据上自如地运行复杂的图算法。

基本词频统计案例

from pyspark import SparkConf, SparkContext
import os

# 绑定指定的Python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    print("Spark入门案例: WordCount词频统计")

    # 1- 创建SparkContext对象
    """
        setAppName:设置PySpark程序运行时的名称
        setMaster:设置PySpark程序运行时的集群模式
    """
    conf = SparkConf()\
        .setAppName('spark_wordcount_demo')\
        .setMaster('local[*]')
    sc = SparkContext(conf=conf)

    # 2- 数据输入
    """
        textFile:支持读取HDFS文件系统和本地文件系统
            HDFS文件系统:hdfs://node1:8020/文件路径
            本地文件系统:file:///文件路径
    """
    init_rdd = sc.textFile("file:///export/data/gz16_pyspark/01_spark_core/data/content.txt")

    # 3- 数据处理
    # 文本内容切分
    """
        flatMap运行结果:
        输入数据:['hello hello spark', 'hello heima spark']
        输出数据:['hello', 'hello', 'spark', 'hello', 'heima', 'spark']
        
        map运行结果:
        输入数据:['hello hello spark', 'hello heima spark']
        输出数据:[['hello', 'hello', 'spark'], ['hello', 'heima', 'spark']]
    """
    # flatmap_rdd = init_rdd.map(lambda line: line.split(" "))
    flatmap_rdd = init_rdd.flatMap(lambda line: line.split(" "))

    # 数据格式转换
    """
        输入数据:['hello', 'hello', 'spark', 'hello', 'heima', 'spark']
        输出数据:[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
    """
    map_rdd = flatmap_rdd.map(lambda word: (word,1))

    # 分组和聚合
    """
        输入数据:[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
        输出数据:[('hello', 3), ('spark', 2), ('heima', 1)]
        
        reduceByKey底层运行过程分析:
        1- 该算子同时具备分组和聚合的功能。而且是先对数据按照key进行分组,对相同key的value会形成得到List列表。再对分组后的value列表进行聚合。
        2- 分组和聚合功能不能分割,也就是一个整体
        
        结合案例进行详细分析:
        1- 分组
            输入数据:[('hello', 1), ('hello', 1), ('spark', 1), ('hello', 1), ('heima', 1), ('spark', 1)]
            分组后的结果: 
                        key  value列表
                      hello  [1,1,1]
                      spark  [1,1]
                      heima  [1]
                      
        2- 聚合(以hello为例)
            lambda agg,curr: agg+curr -> agg表示中间临时value聚合结果,默认取列表中的第一个元素;curr表示当前遍历到的value元素,默认取列表中的第二个元素
            
            第一次聚合:
                agg =列表中的第一个1,curr=列表中的第二个1。聚合结果agg+curr=1+1=2,再将2赋值给agg
            第二次聚合:
                agg =上次临时聚合结果2,curr=列表中的第三个1。聚合结果agg+curr=2+1=3,再将3赋值给agg
                
            最后发现已经遍历到value列表的最后一个元素,因此聚合过程结果。最终的hello的次数,就是3
    """
    result = map_rdd.reduceByKey(lambda agg,curr: agg+curr)

    # 4- 数据输出
    """
        collect():用来收集数据,返回值类型是List列表
    """
    print(result.collect())

    # 5- 释放资源
    sc.stop()

可能出现的错误:
在这里插入图片描述
原因: 找不到JAVA_HOME环境

解决方案: 需要在代码中指定远端的环境地址 以及 在node1环境中初始化JAVA_HOME地址
第一步: 在mian函数上面添加以下内容
os.environ[‘SPARK_HOME’] = ‘/export/server/spark’
os.environ[‘PYSPARK_PYTHON’] = ‘/root/anaconda3/bin/python3’
os.environ[‘PYSPARK_DRIVER_PYTHON’] = ‘/root/anaconda3/bin/python3’
第二步:在node1的 /root/.bashrc 中配置初始化环境的配置
export JAVA_HOME=/export/server/jdk1.8.0_241
注意: 路径需要填写你自己

文章来源:https://blog.csdn.net/qq_50215015/article/details/135445988
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。