Spark学习笔记之Spark中的RDD的具体使用

1. Spark中的RDD

  • Resilient Distributed Datasets(弹性分布式数据集)
  • Spark中的最基本的抽象
  • 有了RDD的存在我们就可以像操作本地集合一样操作分布式的数据
  • 包含所有元素的分区的集合
  • RDD包含了很多的分区
  • 2. RDD中的弹性
  • RDD中的数据是可大可小的
  • RDD的数据默认情况下存放在内存中的,但是在内存资源不足时,Spark会自动将RDD数据写入磁盘
  • RDD有自动容错功能,当其中一个RDD中的分区的数据丢失,或者当前节点故障时,rdd会根据依赖关系重新计算该分区的数据

3. RDD在Spark中的作用

迭代式计算

其主要实现思想就是RDD,把所有计算的数据保存在分布式的内存中。迭代计算通常情况下都是对同一个数据集做反复的迭代计算,数据在内存中将大大提升IO操作。这也是Spark涉及的核心:内存计算

交互式计算

因为Spark是用scala语言实现的,Spark和scala能够紧密的集成,所以Spark可以完美的运用scala的解释器,使得其中的scala可以向操作本地集合对象一样轻松操作分布式数据集

4. Spark中的名词解释

  • ClusterManager :在Standalone模式中即为Master(主节点),控制整个集群,监控Worker。在YARN模式中为资源管理器
  • Worker:从节点,负责控制计算节点,启动Executor。在YARN模式中为NodeManager,负责计算节点的控制。
  • Driver 运行Application的main()函数并创建SparkContext
  • Executor (CoarseGrainedExecutorBackend)在worker node上执行任务的组件、用于启动线程池运行任务。每个Application拥有独立的一组Executors
  • SparkContext :整个应用的上下文,控制应用的生命周期
  • RDD :Spark中的最基本的数据抽象
  • DAG Scheduler : 根据DAG(有向无环图)切分stage,并且生成task,以taskset的形式返回
  • Task Schedual: 调度task,把task交给executor
  • Stage: 一个Spark作业一般包含一到多个Stage。
  • Task :一个Stage包含一到多个Task,通过多个Task实现并行运行的功能
  • Transformations :转换操作,Transformation是lazy的,不会马上执行,只有当调用action时才会执行
  • Actions : 动作
  • SparkEnv : 线程级别的上下文,存储运行时的重要组件的引用

5. 创建RDD的两种方式

通过并行化集合创建RDD(用于测试)

val list = List("java c++ java","java java java c++")
val rdd = sc.parallelize(list)

通过加载hdfs中的数据创建RDD(生产环境)

val rdd = sc.textFile("hdfs://uplooking01:8020/sparktest/")

6. IDEA开发Spark

6.1 pom依赖

<?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">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.uplooking.bigdata</groupId>
  <artifactId>2018-11-08-spark</artifactId>
  <version>1.0-SNAPSHOT</version>

  <properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <scala.version>2.11.8</scala.version>
    <spark.version>2.2.0</spark.version>
    <hadoop.version>2.7.5</hadoop.version>
  </properties>

  <dependencies>
    <!-- 导入scala的依赖 -->
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>${scala.version}</version>
    </dependency>

    <!-- 导入spark的依赖 -->
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.11</artifactId>
      <version>${spark.version}</version>
    </dependency>
    <!-- 指定hadoop-client API的版本 -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>${hadoop.version}</version>
    </dependency>

  </dependencies>

  <build>
    <plugins>
      <!--编译Scala-->
      <plugin>
        <groupId>net.alchim31.maven</groupId>
        <artifactId>scala-maven-plugin</artifactId>
        <version>3.2.2</version>
        <executions>
          <execution>
            <id>scala-compile-first</id>
            <phase>process-resources</phase>
            <goals>
              <goal>add-source</goal>
              <goal>compile</goal>
            </goals>
          </execution>
          <execution>
            <id>scala-test-compile</id>
            <phase>process-test-resources</phase>
            <goals>
              <goal>testCompile</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
      <!--编译Java-->
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <executions>
          <execution>
            <phase>compile</phase>
            <goals>
              <goal>compile</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
      <!-- 打jar插件 -->
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>2.4.3</version>
        <configuration>
          <createDependencyReducedPom>false</createDependencyReducedPom>
        </configuration>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <filters>
                <filter>
                  <artifact>*:*</artifact>
                  <excludes>
                    <exclude>META-INF/*.SF</exclude>
                    <exclude>META-INF/*.DSA</exclude>
                    <exclude>META-INF/*.RSA</exclude>
                  </excludes>
                </filter>
              </filters>
            </configuration>
          </execution>
        </executions>
      </plugin>

    </plugins>
  </build>
</project>

6.2 编写spark程序

val conf = new SparkConf()
conf.setAppName("Ops1")
val sc = new SparkContext(conf)
val rdd1: RDD[String] = sc.parallelize(List("java c+ java", "java java c++"))
val ret = rdd1.collect().toBuffer
println(ret)

6.3 打包

6.4 在Driver上运行jar包

spark-submit --master spark://uplooking01:7077 --class com.uplooking.bigdata.spark01.Ops1 original-spark-1.0-SNAPSHOT.jar

7. 本地运行Spark程序

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

import scala.collection.mutable

object Ops1 {
 def main(args: Array[String]): Unit = {
  val conf = new SparkConf()
  conf.setAppName("Ops1")
  conf.setMaster("local[4]")
  val sc = new SparkContext(conf)
  //一般不会指定最小分区数
  val rdd1 = sc.textFile("hdfs://uplooking01:8020/sparktest/")
  val rdd2: RDD[String] = rdd1.flatMap(line => line.split(" "))
  val rdd3: RDD[(String, Int)] = rdd2.map(word => (word, 1))
  val rdd4: RDD[(String, Int)] = rdd3.reduceByKey(_ + _)
  val ret: mutable.Buffer[(String, Int)] = rdd4.collect().toBuffer
  println(ret)
  println(rdd1.partitions.length)
 }
}

8. RDD中的分区数

并行化的方式指定分区数(一般会指定分区数)

  • 默认如果创建RDD时不指定分区数,那么就会创建cpu核数个分区
  • 手动指定分区数
val rdd = sc.parallelize(List("java c+ java", "java java c++"), 2)

textFile的方式指定分区数

  • 默认如果创建RDD时不指定最小分区数,那么就会创建至少2个分区的RDD
  • 一般不会指定最小分区数
  • 不指定最小分区数,有切片的数量个分区

9. Spark作业的运行流程

  • 构建DAG
  • 根据DAG切分Stage,每个Stage对应一组相同计算逻辑不能计算数据的Task,以TastSet的形式返回
  • TaskSchedual调度task,把task发送到executor中去,用Runnable进行包装进给线程池
  • Executor执行task

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • 将string类型的数据类型转换为spark rdd时报错的解决方法

    在将string类型的数据类型转换为spark rdd时,一直报这个错,StructType can not accept object %r in type %s" % (obj, type(obj))) . . . s = str(tree) y = str(YESTERDAY) list0 = [s, y] outRes = self.sc.parallelize(list0) df_tree = outRes.toDF("model: string, dt: string&qu

  • Java和scala实现 Spark RDD转换成DataFrame的两种方法小结

    一:准备数据源 在项目下新建一个student.txt文件,里面的内容为: 1,zhangsan,20 2,lisi,21 3,wanger,19 4,fangliu,18 二:实现 Java版: 1.首先新建一个student的Bean对象,实现序列化和toString()方法,具体代码如下: package com.cxd.sql; import java.io.Serializable; @SuppressWarnings("serial") public class Stude

  • spark rdd转dataframe 写入mysql的实例讲解

    dataframe是在spark1.3.0中推出的新的api,这让spark具备了处理大规模结构化数据的能力,在比原有的RDD转化方式易用的前提下,据说计算性能更还快了两倍.spark在离线批处理或者实时计算中都可以将rdd转成dataframe进而通过简单的sql命令对数据进行操作,对于熟悉sql的人来说在转换和过滤过程很方便,甚至可以有更高层次的应用,比如在实时这一块,传入kafka的topic名称和sql语句,后台读取自己配置好的内容字段反射成一个class并利用出入的sql对实时数据进行

  • 浅谈Spark RDD API中的Map和Reduce

    RDD是什么? RDD是Spark中的抽象数据结构类型,任何数据在Spark中都被表示为RDD.从编程的角度来看,RDD可以简单看成是一个数组.和普通数组的区别是,RDD中的数据是分区存储的,这样不同分区的数据就可以分布在不同的机器上,同时可以被并行处理.因此,Spark应用程序所做的无非是把需要处理的数据转换为RDD,然后对RDD进行一系列的变换和操作从而得到结果.本文为第一部分,将介绍Spark RDD中与Map和Reduce相关的API中. 如何创建RDD? RDD可以从普通数组创建出来,

  • spark: RDD与DataFrame之间的相互转换方法

    DataFrame是一个组织成命名列的数据集.它在概念上等同于关系数据库中的表或R/Python中的数据框架,但其经过了优化.DataFrames可以从各种各样的源构建,例如:结构化数据文件,Hive中的表,外部数据库或现有RDD. DataFrame API 可以被Scala,Java,Python和R调用. 在Scala和Java中,DataFrame由Rows的数据集表示. 在Scala API中,DataFrame只是一个类型别名Dataset[Row].而在Java API中,用户需要

  • Spark学习笔记之Spark中的RDD的具体使用

    1. Spark中的RDD Resilient Distributed Datasets(弹性分布式数据集) Spark中的最基本的抽象 有了RDD的存在我们就可以像操作本地集合一样操作分布式的数据 包含所有元素的分区的集合 RDD包含了很多的分区 2. RDD中的弹性 RDD中的数据是可大可小的 RDD的数据默认情况下存放在内存中的,但是在内存资源不足时,Spark会自动将RDD数据写入磁盘 RDD有自动容错功能,当其中一个RDD中的分区的数据丢失,或者当前节点故障时,rdd会根据依赖关系重新

  • Spark学习笔记之Spark SQL的具体使用

    1. Spark SQL是什么? 处理结构化数据的一个spark的模块 它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用 2. Spark SQL的特点 多语言的接口支持(java python scala) 统一的数据访问 完全兼容hive 支持标准的连接 3. 为什么学习SparkSQL? 我们已经学习了Hive,它是将Hive SQL转换成MapReduce然后提交到集群上执行,大大简化了编写MapReduce的程序的复杂性,由于MapReduce这种计算模型执

  • ng2学习笔记之bootstrap中的component使用教程

     前序: 现在angular2已经除了集成的angular-cli,建议大家可以基于这个来快速开发ng2的项目,不用自己再搭建环境: 相关内容请前往:https://angular.cn/docs/ts/latest/cli-quickstart.html 正文: 以图片轮播组件carousel为例: 首先安装好ng2-bootstrap,bootstrap npm install ng2-bootstrap bootstrap --save 在需要用到的模块中导入 import { Carou

  • Python GUI编程学习笔记之tkinter中messagebox、filedialog控件用法详解

    本文实例讲述了Python GUI编程学习笔记之tkinter中messagebox.filedialog控件用法.分享给大家供大家参考,具体如下: 相关内容: messagebox 介绍 使用 filedialog 介绍 使用 首发时间:2018-03-04 22:18 messagebox: 介绍:messagebox是tkinter中的消息框.对话框 使用: 导入模块:import tkinter.messagebox 选择消息框的模式: 提示消息框:[返回"ok"] tkint

  • Spark学习笔记 (二)Spark2.3 HA集群的分布式安装图文详解

    本文实例讲述了Spark2.3 HA集群的分布式安装.分享给大家供大家参考,具体如下: 一.下载Spark安装包 1.从官网下载 http://spark.apache.org/downloads.html 2.从微软的镜像站下载 http://mirrors.hust.edu.cn/apache/ 3.从清华的镜像站下载 https://mirrors.tuna.tsinghua.edu.cn/apache/ 二.安装基础 1.Java8安装成功 2.zookeeper安装成功 3.hadoo

  • Spark学习笔记(一)Spark初识【特性、组成、应用】

    本文实例讲述了Spark基本特性.组成.应用.分享给大家供大家参考,具体如下: 一.官网介绍 1.什么是Spark 官网地址:http://spark.apache.org/ Apache Spark™是用于大规模数据处理的统一分析引擎. 从右侧最后一条新闻看,Spark也用于AI人工智能 spark是一个实现快速通用的集群计算平台.它是由加州大学伯克利分校AMP实验室 开发的通用内存并行计算框架,用来构建大型的.低延迟的数据分析应用程序.它扩展了广泛使用的MapReduce计算模型.高效的支撑

  • jQuery学习笔记[1] jQuery中的DOM操作

    DOM分为3个方面,即DOM Core,HTML-DOM,CSS-DOM. 1,DOM Core并不专属于JavaScript,任何一种支持DOM的程序设计语言都可以使用它. JavaScript中的getElementById(),getElementsByTagName()...等方法. 例如:使用DOM Core来获取表单对象的方法: document.getElementsByTagName("form"); 2,HTML-DOM. 在使用JavaScript和DOM为HTML

  • Spark学习笔记Spark Streaming的使用

    1. Spark Streaming Spark Streaming是一个基于Spark Core之上的实时计算框架,可以从很多数据源消费数据并对数据进行处理 Spark Streaing中有一个最基本的抽象叫DStream(代理),本质上就是一系列连续的RDD,DStream其实就是对RDD的封装 DStream可以认为是一个RDD的工厂,该DStream里面生产都是相同业务逻辑的RDD,只不过是RDD里面要读取数据的不相同 在一个批次的处理时间间隔里, DStream只产生一个RDD DSt

  • Vue 2.0学习笔记之Vue中的computed属性

    Vue中的 computed 属性称为 计算属性 .在这一节中,我们学习Vue中的计算属性如何使用?记得在学习Vue的模板相关的知识的时候,知道在模板内可以使用表达式,而且模板内的表达式是非常的便利,但这种遍历是有一定的限制的,它们实际上是用于一些简单的运算.也就是说,如果在模板中放入太多的逻辑会让模板过重而且难以维护.咱们先来看一个示例: <div id="app"> <h1>{{ message.split('').reverse().join('') }}

  • C#.NET学习笔记5 C#中的条件编译

    条件编译是C#比Java多出的东西,但我跟前辈请教后,他们都说条件编译在实际的项目开发中不怎么使用.鉴于是新内容,我还是做做笔记,理解一下好了. 条件编译属于编译预处理的范畴,它能让我们通过条件编译的机制,将部分代码包括进来或者排除出去,其作用与if-else类似.条件编译指令有以下四种

随机推荐