代码驿站移动版

当前位置：主页 > 软件编程 > JAVA代码 >

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

时间：2022-12-01 10:56:16 | 栏目：JAVA代码 | 点击：次

1. Spark中的RDD

3. RDD在Spark中的作用

迭代式计算

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

交互式计算

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

4. Spark中的名词解释

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中的分区数

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

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

textFile的方式指定分区数

9. Spark作业的运行流程

您可能感兴趣的文章:

相关文章