Spark 基础操作

1. Spark 基础

2. Spark Core

3. Spark SQL

4. Spark Streaming

5. Spark 内核机制

6. Spark 性能调优

1. Spark 基础

1.1 Spark 中的相应组件

1.2 Standalone 模式安装

// 1. 准备安装包(见下方参考资料): spark-2.1.3-bin-hadoop2.7.tgz

// 2. 修改配置文件
// 2.1 spark-env.sh.template
mv spark-env.sh.template spark-env.sh
SPARK_MASTER_HOST=IP地址
SPARK_MASTER_PORT=7077

// 3. 启动
sbin/start-all.sh

// 4. 浏览器访问
http://IP地址:8080

// 5. 测试官方案例
bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://IP地址:7077 --executor-memory 1G --total-executor-cores 2 ./examples/jars/spark-examples_2.11-2.1.3.jar 100

// 6. 使用 Spark Shell 测试 WordCount
bin/spark-shell --master spark://10.110.147.193:7077

sc.textFile("./RELEASE").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

1.2.1 提交应用程序概述

1. bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://IP地址:7077 --executor-memory 1G --total-executor-cores 2 ./examples/jars/spark-examples_2.11-2.1.3.jar 100
   • --class: 应用程序的启动类,例如,org.apache.spark.examples.SparkPi;
   • --master: 集群的master URL;
   • deploy-mode: 是否发布你的驱动到worker节点(cluster)或者作为一个本地客户端(client);
   • --conf: 任意的Spark配置属性,格式:key=value,如果值包含空格,可以加引号"key=value";
   • application-jar:打包好的应用 jar,包含依赖,这个URL在集群中全局可见。比如hdfs://共享存储系统,如果是file://path,那么所有节点的path都包含同样的jar;
   • application-arguments: 传给main()方法的参数;

1.3 JobHistoryServer 配置

1. 修改spark-defaults.conf.template名称:mv spark-defaults.conf.template spark-defaults.conf;
2. 修改spark-defaults.conf文件,开启 Log:
   • spark.eventLog.enabled true;
   • spark.eventLog.dir hdfs://IP地址:9000/directory;
   • 注意:HDFS 上的目录需要提前存在;
3. 修改 spark-env.sh 文件,添加如下配置:
   • export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=4000 -Dspark.history.retainedApplications=3 -Dspark.history.fs.logDirectory=hdfs://IP地址:9000/directory";
4. 开启历史服务:sbin/start-history-server.sh;
5. 执行上面的程序:org.apache.spark.examples.SparkPi;
6. 访问:http//IP地址:4000;

1.4 Spark HA 配置

1. zookeeper 正常安装并启动;
2. 修改 spark-env.sh 文件,添加如下配置:
   • 注释掉:
     • SPARK_MASTER_HOST=IP地址;
     • SPARK_MASTER_PORT=7077
   • export SPARK_DEAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=IP地址1, IP地址2, IP地址3 -Dspark.deploy.zookeeper.dir=/spark"

1.5 Yarn 模式安装

1. 修改 hadoop 配置文件yarn-site.xml,添加如下内容:

    yarn.nodemanager.pmem-check-enabled
    false



    yarn.nodemanager.vmem-check-enabled
    false

1. 修改 spark-env.sh,添加如下配置:
   • YARN_CONF_DIR=/opt/module/hadoop-2.8.5/etc/hadoop
   • HADOOP_CONF_DIR=/opt/module/hadoop-2.8.5/etc/hadoop

1.6 Spark Shell WordCount 流程

• sc.textFile("文件地址").flatMap(_.split("")).map((_,1)).reduceByKey(_+_).saveAsTextFile.("输出结果")

1.6 Eclipse 编写 WordCount 程序

// ===== 1. 创建 Maven Project
// ===== 2. 导入依赖 pom.xml

      
          org.apache.spark
          spark-core_2.11
          2.1.3
      
  
  
        WordCount
        
        
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.6.0
                
                    1.8
                    1.8
                    UTF-8
                
            
             
                org.apache.maven.plugins
                maven-surefire-plugin
                2.17
            
            
                net.alchim31.maven
                scala-maven-plugin
                3.2.2
                
                    
                        
                            compile
                            test-compile
                        
                    
                
            
            
                org.apache.maven.plugins
                maven-assembly-plugin
                3.0.0
                
                    
                        
                            com.noodles.WordCount
                        
                    
                    jar-with-dependencies
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    


// ===== 3. 创建 Maven Module
// ===== 4. 创建 Scala Class: WordCount
object WordCount {

    def main(args:Array[String]): Unit = {

        // 1. 创建配置信息
        val conf = new SparkConf().setAppName("wc")

        // 2. 创建 sparkcontext
        val sc = new SparkContext(conf)

        // 3. 处理逻辑

        // 读取数据
        val lines = sc.textFile(args(0))
        val words = lines.flatMap(_.split(" "))
        val k2v = words.map((_, 1))
        val result = k2v.reduceByKey(_+_)

        // 输出
        // result.collect()
        // 将输出结果保存到文件
        result.saveAsTextFile(args(1))

        // 4. 关闭连接
        sc.stop()
    }
}

参考资料:

• spark-2.1.3-bin-hadoop2.7.tgz下载
• eclipse的maven、Scala环境搭建