Spark 测试程序计算圆周率



1. 普通模式提交任务

该算法是利用蒙特·卡罗算法求圆周率PI，通过计算机模拟大量的随机数，最终会计算出比较精确的π。



bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://node-1:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
examples/jars/spark-examples_2.11-2.0.2.jar \
20





2. 高可用模式提交任务

在高可用模式下，因为涉及到多个Master，所以对于应用程序的提交就有了一点变化，因为应用程序需要知道当前的Master的IP地址和端口。这种HA方案处理这种情况很简单，只需要在SparkContext指向一个Master列表就可以了， 

如spark://host1:port1,host2:port2,host3:port3，应用程序会轮询列表，找到活着的Master。



bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://node-1:7077,node-2:7077,node-3:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
examples/jars/spark-examples_2.11-2.0.2.jar \
10

喜欢就点赞评论+关注吧



感谢阅读，希望能帮助到大家，谢谢大家的支持！

1. 
Spark 的四大组件下面哪个不是 ( )

A.Spark Streaming B Mlib 
C Graphx D Spark R


2.下面哪个端口不是 spark 自带服务的端口 ( )

A.8080 B.4040 C.8090 D.18080


3.spark 1.4 版本的最大变化 ( )

A spark sql Release 版本 B 引入 Spark R 
C DataFrame D支持动态资源分配


4. Spark Job 默认的调度模式 ( )

A FIFO   B FAIR   
C 无   D 运行时指定


5.哪个不是本地模式运行的个条件 ( )

A spark.localExecution.enabled=true   B 显式指定本地运行 C finalStage 无父 Stage D partition默认值


6.下面哪个不是 RDD 的特点 ( )

A. 可分区   B 可序列化   C 可修改   D 可持久化


7. 关于广播变量，下面哪个是错误的 ( )

A 任何函数调用    B 是只读的   C 存储在各个节点    D 存储在磁盘或 HDFS


8. 关于累加器，下面哪个是错误的 ( )

A 支持加法 B 支持数值类型 
C 可并行 D 不支持自定义类型


9.Spark 支持的分布式部署方式中哪个是错误的 ( )

A standalone B spark on mesos  C spark on YARN
D Spark on local


10.Stage 的 Task 的数量由什么决定 ( )
A Partition B JobC Stage
D TaskScheduler


11.下面哪个操作是窄依赖 ( )

A join B filter 
C group D sort


12.下面哪个操作肯定是宽依赖 ( )

A map B flatMap 
C reduceByKey D sample


13.spark 的 master 和 worker 通过什么方式进行通信的？ ( )

A http B nio C nettyD Akka


14 默认的存储级别 ( )

A MEMORY_ONLY B MEMORY_ONLY_SER

C MEMORY_AND_DISK D MEMORY_AND_DISK_SER


15 spark.deploy.recoveryMode 不支持那种 ( )

A.ZooKeeper B. FileSystem 
D NONE D 
Hadoop


16.下列哪个不是 RDD 的缓存方法 ( )

A persist() B Cache() 
C Memory()


17.Task 运行在下来哪里个选项中 Executor 上的工作单元 ( )

A Driver program B. spark master 
C.worker node D Cluster manager


18.Hive 的元数据存储在 derby 和

MySQL 中有什么区别 ( )

A.没区别 B.多会话 C.支持网络环境D数据库的区别


19.DataFrame 和 RDD 最大的区别 ( )

A.科学统计支持 B.多了 schema 
C.存储方式不一样 D.外部数据源支持


20.Master 的 ElectedLeader 事件后做了哪些操作 ( )

A. 通知 driver B.通知 worker 
C.注册 application D.直接 ALIVE



答案：
DCBAD   CDDDA
BCDAD   CCBBD

1. spark 的四大组件下面哪个不是 ( )

A.Spark Streaming B Mlib 
C Graphx D Spark R


2.下面哪个端口不是 spark 自带服务的端口 ( )

A.8080 B.4040 C.8090 D.18080


3.spark 1.4 版本的最大变化 ( )

A spark sql Release 版本 B 引入 Spark R 
C DataFrame D支持动态资源分配


4. Spark Job 默认的调度模式 ( )

A FIFO   B FAIR   
C 无   D 运行时指定


5.哪个不是本地模式运行的个条件 ( )

A spark.localExecution.enabled=true   B 显式指定本地运行 C finalStage 无父 Stage D partition默认值


6.下面哪个不是 RDD 的特点 ( )

A. 可分区   B 可序列化   C 可修改   D 可持久化


7. 关于广播变量，下面哪个是错误的 ( )

A 任何函数调用    B 是只读的   C 存储在各个节点    D 存储在磁盘或 HDFS


8. 关于累加器，下面哪个是错误的 ( )

A 支持加法 B 支持数值类型 
C 可并行 D 不支持自定义类型


9.Spark 支持的分布式部署方式中哪个是错误的 ( )

A standalone B spark on mesos  
C spark on YARN D Spark on local


10.Stage 的 Task 的数量由什么决定 ( )
A Partition B Job
C Stage D TaskScheduler


11.下面哪个操作是窄依赖 ( )

A join B filter 
C group D sort


12.下面哪个操作肯定是宽依赖 ( )

A map B flatMap 
C reduceByKey D sample


13.spark 的 master 和 worker 通过什么方式进行通信的？ ( )

A http B nio C netty
D Akka


14 默认的存储级别 ( )

A MEMORY_ONLY B MEMORY_ONLY_SER

C MEMORY_AND_DISK D MEMORY_AND_DISK_SER


15 spark.deploy.recoveryMode 不支持那种 ( )

A.ZooKeeper B. FileSystem 
D NONE D hadoop


16.下列哪个不是 RDD 的缓存方法 ( )

A persist() B Cache() 
C Memory()


17.Task 运行在下来哪里个选项中 Executor 上的工作单元 ( )

A Driver program B. spark master 
C.worker node D Cluster manager


18.hive 的元数据存储在 derby 和 mysql 中有什么区别 ( )

A.没区别 B.多会话 C.支持网络环境
D数据库的区别


19.DataFrame 和 RDD 最大的区别 ( )

A.科学统计支持 B.多了 schema 
C.存储方式不一样 D.外部数据源支持


20.Master 的 ElectedLeader 事件后做了哪些操作 ( )

A. 通知 driver B.通知 worker 
C.注册 application D.直接 ALIVE



答案：
DCBAD   CDDDA
BCDAD   CCBBD

使用scalaTest工具,用法参考: 
scalaTest的使用



代码

src/test/tool/LocalSparkContext.scala



import org.apache.spark.{SparkConf, SparkContext}
import org.scalatest._

trait LocalSparkContext extends BeforeAndAfterAll {
  self: Suite =>

  @transient var sc: SparkContext = _

  override def beforeAll() {
    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("test")
    sc = new SparkContext(conf)
  }

  override def afterAll() {
    if (sc != null) {
      sc.stop()
    }
  }
}

src/test/SparkWCSuit



import org.apache.spark.sql.{Row, SQLContext}
import org.scalatest.FunSuite
import tool.LocalSparkContext

class SparkWCSuit extends FunSuite
  with LocalSparkContext {

//rdd wordCount
  test("test rdd wc") {
    sc.setLogLevel("ERROR")
    val rdd = sc.makeRDD(Seq("a", "b", "b"))
    val res = rdd.map((_, 1)).reduceByKey(_ + _).collect().sorted
    assert(res === Array(("a", 1), ("b", 2)))
  }

//df wordCount
  test("test df wc") {
    val sqlContext = SQLContext.getOrCreate(sc)
    import sqlContext.implicits._
    val df = sc.makeRDD(Seq("a", "b", "b")).toDF("word")
    val res = df.groupBy("word").count().collect()
    assert(res === Array(Row("a",1),Row("b",2)))
  }
}



结果