好程序员大数据教程： SparkShell 和 IDEA 中编写 Spark 程序， spark-shell是Spark自带的交互式Shell程序，方便用户进行交互式编程，用户可以在该命令行下用Scala编写Spark程序。spark-shell程序一般用作Spark程序测试练习来用。spark-shell属于Spark的特殊应用程序，我们可以在这个特殊的应用程序中提交应用程序

spark-shell启动有两种模式，local模式和cluster模式，分别为

local 模式 :

spark-shell

local模式仅在本机启动一个SparkSubmit进程，没有与集群建立联系,虽然进程中有SparkSubmit但是不会被提交到集群红

Cluster 模式 ( 集群模式 ):

spark-shell \
--master spark://hadoop01:7077 \
--executor-memory 512m \
--total-executor-cores 1

后两个命令不是必须的  --master 这条命令是必须的 ( 除非在 jar 包中已经指可以不指定 , 不然就必须指定 )

退出 shell

千万不要ctrl+c spark-shell 正确退出 :quit 千万不要ctrl+c退出 这样是错误的 若使用了ctrl+c退出 使用命令查看监听端口 netstat - apn | grep 4040 在使用kill -9 端口号 杀死即可

3.25.11 spark2.2shell 和 spark1.6shell 对比

ps:启动spark-shell若是集群模式,在webUI会有一个一直执行的任务

通过 IDEA 创建 Spark 工程

ps:工程创建之前步骤省略,在scala中已经讲解,直接默认是创建好工程的

对工程中的 pom.xml 文件配置

  <!--  声明公有的属性  -->
<properties>
         <maven.compiler.source> 1.8 </maven.compiler.source>
         <maven.compiler.target> 1.8 </maven.compiler.target>
         <encoding> UTF-8 </encoding>
         <scala.version> 2.11.8 </scala.version>
         <spark.version> 2.2.0 </spark.version>
         <hadoop.version> 2.7.1 </hadoop.version>
         <scala.compat.version> 2.11 </scala.compat.version>
     </properties>
<!--  声明并引入公有的依赖  -->
     <dependencies>
         <dependency>
             <groupId> org.scala-lang </groupId>
             <artifactId> scala-library </artifactId>
             <version> ${scala.version} </version>
         </dependency>
         <dependency>
         <groupId> org.apache.spark </groupId>
         <artifactId> spark-core_2.11 </artifactId>
         <version> ${spark.version} </version>
     </dependency>
         <dependency>
             <groupId> org.apache.hadoop </groupId>
             <artifactId> hadoop-client </artifactId>
             <version> ${hadoop.version} </version>
         </dependency>
     </dependencies>  

Spark实现 WordCount 程序

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

object  SparkWordCount {
   def   main (args: Array[String]): Unit = {
     val  conf =  new   SparkConf (). setAppName ( " dri/wordcount " ) .setMaster( " local[*] " )
     // 创建 sparkContext 对象
     val  sc =   new   SparkContext (conf)
     // 通过 sparkcontext 对象就可以处理数据
     // 读取文件 参数是一个 String 类型的字符串 传入的是路径
     val  lines: RDD[String] = sc. textFile ( “ dir/wordcount ” )
     // 切分数据
     val  words: RDD[String] = lines. flatMap (_. split ( " " ))
     // 将每一个单词生成元组  ( 单词 ,1)
     val  tuples: RDD[(String, Int)] = words. map ((_, 1 ))
     //spark 中提供一个算子  reduceByKey  相同 key  为一组进行求和 计算 value
     val  sumed: RDD[(String, Int)] = tuples. reduceByKey (_+_)
     // 对当前这个结果进行排序  sortBy  和 scala 中 sotrBy 是不一样的 多了一个参数
     // 默认是升序   false 就是降序
     val  sorted: RDD[(String, Int)] = sumed. sortBy (_._ 2 , false )
     // 将数据提交到集群存储 无法返回值
       sorted.foreach(println)
     // 回收资源停止 sc, 结束任务
    sc. stop ()
  }
}

Java 版本

import  org. apache . spark . SparkConf ;
import  org. apache . spark . api . java . JavaPairRDD ;
import  org. apache . spark . api . java . JavaRDD ;
import  org. apache . spark . api . java . JavaSparkContext ;
import  org. apache . spark . api . java . function . FlatMapFunction ;
import  org. apache . spark . api . java . function . Function2 ;
import  org. apache . spark . api . java . function . PairFunction ;
import  scala. Tuple2 ;
import  java. util . Arrays ;
import  java. util . Iterator ;
import  java. util . List ;  
public  class  JavaWordCount {
    public static void  main (String[] args) {
//1. 先创建 conf 对象进行配置主要是设置名称 , 为了设置运行模式
        SparkConf conf =  new   SparkConf (). setAppName ( "JavaWordCount" ). setMaster ( "local" );
//2. 创建 context 对象
        JavaSparkContext jsc =  new   JavaSparkContext (conf);
        JavaRDD<String> lines = jsc. textFile ( "dir/file" );
// 进行切分数据  flatMapFunction 是具体实现类
        JavaRDD<String> words = lines. flatMap ( new  FlatMapFunction<String, String>() {
            @Override
            public Iterator<String>  call (String s) throws Exception {
                List<String> splited = Arrays. asList (s. split ( " " ));
                 return  splited. iterator ();
            }

        });
// 将数据生成元组
// 第一个泛型是输入的数据类型 后两个参数是输出参数元组的数据
        JavaPairRDD<String, Integer> tuples = words. mapToPair ( new  PairFunction<String, String,
                Integer>() {
            @Override
            public Tuple2<String, Integer>  call (String s) throws Exception {
                 return   new  Tuple2<String, Integer>(s,  1 );
            }
        });
// 聚合
        JavaPairRDD<String, Integer> sumed = tuples. reduceByKey ( new  Function2<Integer, Integer,
                Integer>() {
            @Override
// 第一个 Integer 是相同 key 对应的 value
// 第二个 Integer 是相同 key  对应的 value
            public Integer  call (Integer v1, Integer v2) throws Exception {
                 return  v1 + v2;
            }
        });
// 因为 Java api 没有提供 sortBy 算子 , 此时需要将元组中的数据进行位置调换 , 然后在排序 , 排完序在换回
// 第一次交换是为了排序
        JavaPairRDD<Integer, String> swaped = sumed. mapToPair ( new  PairFunction<Tuple2<String,
                Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String>  call (Tuple2<String, Integer> tup) throws Exception {
                 return  tup. swap ();
            }
        });
// 排序
        JavaPairRDD<Integer, String> sorted = swaped. sortByKey ( false );
// 第二次交换是为了最终结果  < 单词 , 数量 >
        JavaPairRDD<String, Integer> res = sorted. mapToPair ( new  PairFunction<Tuple2<Integer,
                String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer>  call (Tuple2<Integer, String> tuple2) throws Exception
            {
                 return  tuple2. swap ();
            }
        });
        System. out . println (res. collect ());
        res. saveAsTextFile ( "out1" );
        jsc. stop ();
    }
}