2019-02-17 09:33:59 boling_cavalry 阅读数 1766
  • IDEA开发工具+Maven使用详解视频课程(适合初学者的...

    本课程从IDEA开发工具的安装及配置使用讲起,详细讲解Maven项目管理工具,适合初学者的教程,让你少走弯路! 1.Maven简介及安装 2.配置本地仓库和镜像仓库、项目的目录结构 3.pom.xml详解 4.继承、聚合、依赖 5.使用Nexus搭建私服 6.Maven综合实战应用 教学全程采用笔记+代码案例的形式讲解,通俗易懂!!!

    3034 人正在学习 去看看 汤小洋

Scala语言在函数式编程方面的优势适合Spark应用开发,IDEA是我们常用的IDE工具,今天就来实战IDEA开发Scala版的Spark应用;

版本信息

以下是开发环境:

  1. 操作系统:win10;
  2. JDK:1.8.0_191;
  3. IntelliJ IDEA:2018.2.4 (Ultimate Edition)

以下是运行环境:

  1. Spark:2.3.3;
  2. Scala:2.11.12;
  3. Hadoop:2.7.7;

如果您想在几分钟之内搭建好Spark集群环境,请参考《docker下,极速搭建spark集群(含hdfs集群)》

下载Spark安装包

  1. 去spark官网下载spark安装包,里面有开发时所需的库,如下图,地址是:http://spark.apache.org/downloads.html
    在这里插入图片描述
  2. 将下载好的文件解压,例如我这里解压后所在目录是:C:\software\spark-2.3.3-bin-hadoop2.7

IDEA安装scala插件

  1. 打开IDEA,选择"Configure"->“Plugins”,如下图:
    在这里插入图片描述
  2. 如下图,在红框1输入"scala",点击红框2,开始在中央仓库说搜索:
    在这里插入图片描述
  3. 在搜索结果中选中"scala",再点击右侧的"Install",如下:
    在这里插入图片描述
  4. 等待在线安装成功后,点击"Restart IntelliJ IDEA",如下:
    在这里插入图片描述

新建scala工程

  1. 点击下图红框,创建一个新工程:
    在这里插入图片描述
  2. 在弹出窗口中选择"Scala"->“IDEA”,如下图:
    在这里插入图片描述
  3. 如下图,在红框1中输入项目名称,点击红框2,选择Scala SDK:
    在这里插入图片描述
  4. 在弹出的窗口选择"2.11.12"版本,如下图:
    在这里插入图片描述
  5. 点击下图红中的"Finish",完成工程创建:
    在这里插入图片描述
  6. 工程创建成功了,接下来是添加spark库,点击"File"->“Project Structure”,如下图:
    在这里插入图片描述
  7. 在弹出窗口选择新增一个jar库,如下图:
    在这里插入图片描述
  8. 在弹出窗口选择前面安装的spark-2.3.3-bin-hadoop2.7文件夹下面的jar文件夹,如下:
    在这里插入图片描述
  9. 如下图,弹出的窗口用来选择模块,就选工程目录即可:
    在这里插入图片描述
  10. 至此,整个spark开发环境已经设置好了,现在写一个demo试试,创建一个object,源码如下:
package com.bolingcavalry.sparkscalademo.app

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

/**
  * @Description: 第一个scala语言的spark应用
  * @author: willzhao E-mail: zq2599@gmail.com
  * @date: 2019/2/16 20:23
  */
object FirstDemo {
  def main(args: Array[String]): Unit={
    val conf = new SparkConf()
      .setAppName("first spark app(scala)")
      .setMaster("local[1]");

    new SparkContext(conf)
      .parallelize(List(1,2,3,4,5,6))
      .map(x=>x*x)
      .filter(_>10)
      .collect()
      .foreach(println);
  }
}

以上代码的功能很简单:创建用一个数组,将每个元素做平方运算,再丢弃小于10的元素,然后逐个打印出来;
11. 代码完成后,点击右键选择"Run FirstDemo",即可立即在本机运行,如下图:
在这里插入图片描述
12. 由于windows环境并没有做hadoop相关配置,因此控制台上会有错误堆栈输出,但这些信息并不影响程序运行(本例没有用到hadoop),输出如下,可见结果已经被打印出来(16、25、36):

2019-02-17 09:04:21 INFO  TaskSchedulerImpl:54 - Removed TaskSet 0.0, whose tasks have all completed, from pool 
2019-02-17 09:04:21 INFO  DAGScheduler:54 - ResultStage 0 (collect at FirstDemo.scala:20) finished in 0.276 s
2019-02-17 09:04:21 INFO  DAGScheduler:54 - Job 0 finished: collect at FirstDemo.scala:20, took 0.328611 s
16
25
36
2019-02-17 09:04:21 INFO  SparkContext:54 - Invoking stop() from shutdown hook
2019-02-17 09:04:21 INFO  AbstractConnector:318 - Stopped Spark@452ba1db{HTTP/1.1,[http/1.1]}{0.0.0.0:4040}
2019-02-17 09:04:21 INFO  SparkUI:54 - Stopped Spark web UI at http://DESKTOP-82CCEBN:4040
2019-02-17 09:04:21 INFO  MapOutputTrackerMasterEndpoint:54 - MapOutputTrackerMasterEndpoint stopped!
2019-02-17 09:04:21 INFO  MemoryStore:54 - MemoryStore cleared
2019-02-17 09:04:21 INFO  BlockManager:54 - BlockManager stopped
2019-02-17 09:04:21 INFO  BlockManagerMaster:54 - BlockManagerMaster stopped
2019-02-17 09:04:21 INFO  OutputCommitCoordinator$OutputCommitCoordinatorEndpoint:54 - OutputCommitCoordinator stopped!
2019-02-17 09:04:21 INFO  SparkContext:54 - Successfully stopped SparkContext
2019-02-17 09:04:21 INFO  ShutdownHookManager:54 - Shutdown hook called
2019-02-17 09:04:21 INFO  ShutdownHookManager:54 - Deleting directory C:\Users\12167\AppData\Local\Temp\spark-4bbb584a-c7c2-4dc8-9c7e-473de7f8c326

Process finished with exit code 0

构建打包,提交到spark环境运行

  1. 生产环境下一般是将应用构建打包成jar,放到spark集群中运行,所以我们来操作构建打包;
  2. 在菜单上选择"File"->“Project Structure”,弹出窗口点击"Artifacts",选择新增jar,如下图:
    在这里插入图片描述
  3. 如下图,在弹出的窗口中,红框1位置输入要运行的class,红框2选择的是单选框的第二个"copy to the output …":
    在这里插入图片描述
  4. 在菜单上选择"Build"->“Build Artifacts…”,如下图:
    在这里插入图片描述
  5. 在弹出的菜单中选择"sparkscalademo:jar"->“Rebuild”,如下:
    在这里插入图片描述
  6. 如果编译成功,在项目的out\artifacts目录下就会生成文件sparkscalademo.jar,如下:
    在这里插入图片描述
  7. 将文件上传到spark服务器上,执行提交命令:
spark-submit --class com.bolingcavalry.sparkscalademo.app.FirstDemo /root/jars/sparkscalademo.jar
  1. 控制台会显示运行信息和结果,如下图:
    在这里插入图片描述
    至此,idea开发spark应用实战就完成了,希望在您配置开发环境的时候本文能够提供一些参考;

欢迎关注我的公众号:程序员欣宸

在这里插入图片描述

2018-10-16 11:45:38 weixin_42051109 阅读数 917
  • IDEA开发工具+Maven使用详解视频课程(适合初学者的...

    本课程从IDEA开发工具的安装及配置使用讲起,详细讲解Maven项目管理工具,适合初学者的教程,让你少走弯路! 1.Maven简介及安装 2.配置本地仓库和镜像仓库、项目的目录结构 3.pom.xml详解 4.继承、聚合、依赖 5.使用Nexus搭建私服 6.Maven综合实战应用 教学全程采用笔记+代码案例的形式讲解,通俗易懂!!!

    3034 人正在学习 去看看 汤小洋

摘要:

    我们一般选择IntelliJ IDEA书写spark程序,然后打包成jar文件,放到spark集群中运行,接下来我将以WordCount为例仔细讲述Scala程序的 "创建 => 编写 => 打包 => 运行 " 这一过程。

所需工具:

1.  spark 集群(并已经配置好Standalone模式,我的spark集群是在docker中部署的,上一篇博文讲过如何搭建hadoop完全分布式

2.  IntelliJ IDEA 开发环境

          =>配置了 java_1.8

          =>配置了 scala-sdk-2.11.8

3. 浏览器

一、 创建 Scala 文件

1.  如图:我们打开Idea

2. 点击 Create New Projects

3.   选择Java ,再一直点 next

4.  输入Project name ,再点 finish

5.   右键点击项目名称(我的是WordCount),再左键点击Add Framework support ....

6.  添加Maven依赖(用于编译scala,并打包成jar),再点Ok

7.  接下来我们配置Maven,其中gropuId你可以选择这个wordcount项目所在的文件夹,这个你随意,其添加的代码段为:

<?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>IdeaProjects</groupId>
    <artifactId>WordCount</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
    </dependencies>
    
</project>

8.  然后点击Idea界面右方的 Maven管理器,打开之后点击 ”刷新“ 图标

注意:假如你的界面没有Maven管理器,你可以打开 View ,再点击里面的 Tool Buttons ,这样Maven管理器就出来了。

9.  刷新之后,我们会看到Maven管理器中组建了Dependencies依赖,之后我们再右键点击项目文件(我的文件是WordCount),点击Add Framework support...,添加Scala编程框架。

10.  选择Scala,并选择Scala-sdk-2.11.8 。假如你没有,点击Create创建,再点击download下载需要的Scala-sdk版本。

假如你没有scala-sdk-2.11.8 ,按如下操作

11. 添加Scala框架之后,我们就可以新建Scala 类,编写程序了,点开项目文件WordCount,再点开src文件,再点开main文件,找到java资源文件,右键此文件,新建Scala Class

12.  之后我们输入name ,并选择Object类,再点击Ok

13.  进入scala编程界面。在编写程序之前,我们还需要设置一下编译器,使其自动编译,不然后期打包的jar文件是极其不完整的。点击File ->  点击Setting -> 点击Build下的compiler ->  勾选上 Build project automatically -> 点击 apply ->  ok

二、编写程序

14.  程序编写如下:

<1.> 我们先用local模式,在idea中调试,其实假如放到spark集群上跑,就是把local模式和appname注释掉,在spark-submit提交中使用spark://Master:7077模式,请听我娓娓道来......

package IdeaProjects
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
  def main(args: Array[String]): Unit = {
    //创建spark配置对象
    val conf = new SparkConf()
    //设置master属性
    conf.setMaster("local")
    conf.setAppName("WordCount")
    //通过conf创建sc
    val sc = new SparkContext(conf)
    //书写单词计数程序
    val textFile = sc.textFile("hdfs://Master:9000/user/zhangqingfeng/data/test.txt")
    val wordCount = textFile.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    wordCount.foreach(println)
  }
}

注意:这里的package后的名就是你设置Maven时的groupId的名

输出结果(输出的无用信息有点多,你需要一直往下找):

注意:我这里使用的文本测试文件是在hdfs中的,因此,在你开始调试此程序之前,你需要把hadoop集群启动起来。WebUi地址是 http://Master:50070 ,(这里的Master是我的主机ip,你要根据你自己的实际情况来输入)

     <2.> spark集群模式:假如你想要放到spark集群上跑,而不是local测试,那么你只要改三个参数即可,把setMaster,setAppname这两行注释掉,把sc.textFile("hdfs://........")改为sc.textFile(args(0))   如图所示

完整代码如下,敲完之后你就别运行了,肯定报错,因为缺少必要的参数

package IdeaProjects
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
  def main(args: Array[String]): Unit = {
    //创建spark配置对象
    val conf = new SparkConf()
    //设置master属性
    //conf.setMaster("local")
    //conf.setAppName("WordCount")
    //通过conf创建sc
    val sc = new SparkContext(conf)
    //书写单词计数程序
    val textFile = sc.textFile(args(0))
    val wordCount = textFile.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    wordCount.foreach(println)
  }
}

三、程序打包

15.  程序打包成jar,就像是创建scala类有至少三种方法,打包也有至少两种方法,我将介绍用Maven管理器打包

点击右方的Maven管理器 -> 点击Lifecycle -> 右键点击package -> 点击第二项 run WordCount [package]

16. 打包编译结束后,我们在左边文件目录会看到一个 target 目录,点开target目录,右键点击WordCount-1.0-SNAPSHOT.jar ,选择File Path ,之后点击WordCount-1.0-SNAPSHOT.jar  ,我们会进入jar文件的文件夹,双击jar文件,打开IdeaProjects文件,会有生成的六个class文件,说明打包成功。

四、在spark集群(Standalone)中运行jar包

17.  打包之后,我们便开始启动spark集群了,因为每个人集群不一样,尽量把这个jar文件放到spark集群的共享目录下,我的是在docker中部署的spark集群,所以需要把这个jar文件拷贝到docker容器中。

<1.> 启动里spark的standalone模式后,我们用浏览器打开WebUi,地址是http://Master:8080  (注意:每个的Master地址不同,按照你自己的实际情况来输入ip地址),如图:

从图中,小伙伴也看到了,我这是cluster 模式(从第三行可以看出),待会的运行结果输出我们在控制台是看不到的,需要进入WebUi中的stdout中查看。

<2.>  之后我们进入spark的bin文件目录下,开始提交jar包,命令如下:

[root@Master spark-2.3.0-bin-hadoop2.7]# cd bin
[root@Master bin]# spark-submit --master spark://Master:7077 --name MyWordCountScala --class IdeaProjects.WordCount --executor-memory 512M /usr/local/share/WordCount-1.0-SNAPSHOT.jar hdfs://Master:9000/user/zhangqingfeng/data/test.txt

控制台不输出结果,因为从WebUi中看出是cluster模式,如图:

<3. >  我们打开WebUi,点击Completed Applications 中的刚运行的app id,查看每个worker 的 stdout,查看输出结果:

二、在spark-submit提交中会有一些错误,在WebUi中的stderr中也会发现,其原因基本是内存不足导致的,我把我的spark配置分享给大家(基本是运行内存的设置和worker工作数),至于参数调优我也在进一步学习:

<1.> 打开spark-env.sh 文件

<2.> 打开spark-defaults.conf 文件

五、 善意的提示

    使用了Idea开发spark程序,每次都要打成jar包 放到 spark集群或者本地模式,这样太麻烦。你也可以在Idea中右键直接执行,并且使用spark集群模式,不过你需要在程序中加入一个步骤:

     1.  在Idea中右键执行之前,你需要把此项目打成jar包。

     2.  把此jar包的绝对路径添加到程序中,然后再右键执行就可以了。具体你可以看我下面的代码:

   用了  sc.addJar("绝对路径"),如此以来,你就可以在Idea直接跑程序了。

2016-11-14 16:18:31 HelloJFS 阅读数 331
  • IDEA开发工具+Maven使用详解视频课程(适合初学者的...

    本课程从IDEA开发工具的安装及配置使用讲起,详细讲解Maven项目管理工具,适合初学者的教程,让你少走弯路! 1.Maven简介及安装 2.配置本地仓库和镜像仓库、项目的目录结构 3.pom.xml详解 4.继承、聚合、依赖 5.使用Nexus搭建私服 6.Maven综合实战应用 教学全程采用笔记+代码案例的形式讲解,通俗易懂!!!

    3034 人正在学习 去看看 汤小洋

使用IDEA开发spark程序 (windows)

  1. 安装JDK

    配置环境变量(计算机属性 - 高级设置 - 环境变量 - 新建 ):

    JAVA_HOM= E:\Java\jdk1.8.0_101
    PATH=E:\Java\jdk1.8.0_101\bin
    CLASSPATH=E:\Java\jdk1.8.0_101\lib 
    
  2. 安装scala

    这个直接安装就可以了。但是要特别注意版本(会出现spark不兼容的情况),最好选用2.10的版本,我用的是2.10.4

  3. 配置Hadoop

    HADOOP_HOME=E:\hadoop\deploy\hadoop-1.2.1
    PATH=E:\hadoop\deploy\hadoop-1.2.1\bin
    

注意:向Hadoop的bin目录下添加winutils.exe文件。否则会报错:
没有winutile.exe

附下载地址:链接: https://pan.baidu.com/s/1c01FEe 密码: 6e4k

4.配置spark应用开发环境

4.1 安装Intellij IDEA,在 “Configure” -> “Plugin” -> “Browse repositories” -> 输入scala ; (这里如果下载很慢或者失败可以直接去官网下载插件,然后这里选择本地文件)

4.2 用户在Intellij IDEA 中创建Scala Project, SparkTest

4.3 在菜单栏 “File” -> “project structure” -> “Libraries” 命定,单击 “+“,导入
spark-assembly-1.3.0-SNAPSHOT-hadoop2.5.0-cdh5.3.0.jar 包;

附下载地址:

5.试运行SparkPi程序

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

/**
  * Created by Administrator on 2016/11/14.
  */
object Main {
  def main(args: Array[String]):Unit= {
    val conf = new SparkConf().setAppName("Spark Pi").setMaster("local")
                            //在本地运行设置Maser为local或者local[N]
    val spark = new SparkContext(conf)
    val slices = if (args.length > 0) args(0).toInt else 2
    val n = 100000 * slices
    val count = spark.parallelize(1 to n, slices).map { i =>
      val x = Math.random * 2 - 1
      val y = Math.random * 2 - 1
      if (x * x + y * y < 1) 1 else 0
    }.reduce(_ + _)
    println("Pi is rounghly " + 4.0 * count / n)
    spark.stop()
  }
}

运行结果为:

SparkPi运行结果:

注意:不要在Edit configurations里面设置 “Program argument” 为 local,否则会出错;

错误为:

set.Master

2016-07-05 16:41:52 kwu_ganymede 阅读数 25929
  • IDEA开发工具+Maven使用详解视频课程(适合初学者的...

    本课程从IDEA开发工具的安装及配置使用讲起,详细讲解Maven项目管理工具,适合初学者的教程,让你少走弯路! 1.Maven简介及安装 2.配置本地仓库和镜像仓库、项目的目录结构 3.pom.xml详解 4.继承、聚合、依赖 5.使用Nexus搭建私服 6.Maven综合实战应用 教学全程采用笔记+代码案例的形式讲解,通俗易懂!!!

    3034 人正在学习 去看看 汤小洋

基于IntelliJ IDEA开发Spark的Maven项目——Scala语言


1、Maven管理项目在JavaEE普遍使用,开发Spark项目也不例外,而Scala语言开发Spark项目的首选。因此需要构建Maven-Scala项目来开发Spark项目,本文采用的工具是IntelliJ IDEA 2016,IDEA工具越来越被大家认可,开发java, python ,scala 支持都非常好

下载链接 : https://www.jetbrains.com/idea/download/

安装直接下一步即可


2、安装scala插件,File->Settings->Editor->Plugins,搜索scala即可安装



可能由于网络的原因下载不了,可以采取离线安装的方式,例如:


提示下载失败后,根据提示的地址下载离线安装包 http://plugins.jetbrains.com/files/631/24825/python-145.86.zip

在界面选择离线安装即可:



3、创建Maven工程,File->New Project->Maven

选择相应的JDK版本,直接下一步


设定Maven项目的GroupId及ArifactId


创建项目的工程名称,点击完成即可

创建Maven工程完毕,默认是Java的,没关系后面我们再添加scala与spark的依赖



4、修改Maven项目的pom.xml文件,增加scala与spark的依赖

<?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.ganymede</groupId>
    <artifactId>sparkplatformstudy</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spark.version>1.6.0</spark.version>
        <scala.version>2.10</scala.version>
        <hadoop.version>2.6.0</hadoop.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.39</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
    </dependencies>

    <!-- maven官方 http://repo1.maven.org/maven2/  或 http://repo2.maven.org/maven2/ (延迟低一些) -->
    <repositories>
        <repository>
            <id>central</id>
            <name>Maven Repository Switchboard</name>
            <layout>default</layout>
            <url>http://repo2.maven.org/maven2</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>

        <plugins>
            <plugin>
                <!-- MAVEN 编译使用的JDK版本 -->
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.3</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>




5、删除项目的java目录,新建scala并设置源文件夹


添加scala的SDK


添加scala的SDK成功


6、开发Spark实例


测试案例来自spark官网的mllib例子 http://spark.apache.org/docs/latest/mllib-data-types.html

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

/**
  * Created by wuke on 2016/7/5.
  */
object LoadLibSVMFile extends App{
  import org.apache.spark.mllib.regression.LabeledPoint
  import org.apache.spark.mllib.util.MLUtils
  import org.apache.spark.rdd.RDD

  val conf = new SparkConf().setAppName("LogisticRegressionMail").setMaster("local")

  val sc = new SparkContext(conf)
  val examples: RDD[LabeledPoint] = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")

  println(examples.first)
}

测试通过


7、打包编译,线上发布



注意选择依赖包



2017-01-18 12:49:28 a11a2233445566 阅读数 286
  • IDEA开发工具+Maven使用详解视频课程(适合初学者的...

    本课程从IDEA开发工具的安装及配置使用讲起,详细讲解Maven项目管理工具,适合初学者的教程,让你少走弯路! 1.Maven简介及安装 2.配置本地仓库和镜像仓库、项目的目录结构 3.pom.xml详解 4.继承、聚合、依赖 5.使用Nexus搭建私服 6.Maven综合实战应用 教学全程采用笔记+代码案例的形式讲解,通俗易懂!!!

    3034 人正在学习 去看看 汤小洋


 配置开发环境

1. 要在本地安装好javascala

由于spark1.6需要scala 2.10.X版本推荐 2.10.4java版本最好是1.8所以提前我们要需要安装好java和scala 并在环境变量中配置好

2. 下载 IDEA 社区版本选择windows 版本并按照配置。

安装完成以后启动IDEA进行配置,默认即可然后点击ok以后,设置ui风格 然后点击next 会出现插件的选择页面,默认不需求修改,点击next选择安装scala语言,点击 install按钮(非常重要,以为要开发spark程序所以必须安装),等安装完成以后点击start启动 IDEA

3. 创建scala项目

点击  create new project ,然后填写project name 为“Wordcount” 选择项目的保存地址 project location。

然后设置project sdk 即java 的安装目录。点击右侧的new 按钮,选择jdk,然后选择java安装路径即可 

然后选择scala sdk 。点击右侧的create ,默认出现时2.10.x版本scala ,点击ok即可。然后点击finish

4. 设置 spark的jar 依赖

点击file->project structure 来设置工程的libraries核心是添加sparkjar依赖。 选择 Libraries ,点击右侧的加号,选择java选择spark1.6.0 spark-1.6.0-bin-hadoop2.6\lib\spark-assembly-1.6.0-hadoop2.6.0.jar 点击ok稍等片刻后然后点击ok (Libraries作用WordCount),然后点击apply,点击ok(这一步很重要,如果没有无法编写spark的代码)

 编写代码

1. 在src 下建立spark程序工程包

在src上右击 new ->package 填入package 的name为com.dt.spark 。

2. 创建scala的入口

包的名字上右击 选择 new ->scala class 。在弹出框中 填写Name ,并制定kind 为object ,点击ok。

3. 编写local代码

package com.dt.spark

 

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.rdd.RDD

/**

 * 使用Scala开发本地测试的Spark WordCount程序

 * @author DT大数据梦工厂

 * 新浪微博:http://weibo.com/ilovepains/

 */

object WordCount {

    def main(args: Array[String]){

      /**

       * 1步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,

       * 例如说通过setMaster来设置程序要链接的Spark集群的MasterURL,如果设置

       * local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如

       * 只有1G的内存)的初学者       *

       */

      val conf = new SparkConf() //创建SparkConf对象

      conf.setAppName("Wow,My First Spark in IDEA!") //设置应用程序的名称,在程序运行的监控界面可以看到名称

      conf.setMaster("local") //此时,程序在本地运行,不需要安装Spark集群

      

      /**

       * 2步:创建SparkContext对象

       * SparkContextSpark程序所有功能的唯一入口,无论是采用ScalaJavaPythonR等都必须有一个SparkContext

       * SparkContext核心作用:初始化Spark应用程序运行所需要的核心组件,包括DAGSchedulerTaskSchedulerSchedulerBackend

       * 同时还会负责Spark程序往Master注册程序等

       * SparkContext是整个Spark应用程序中最为至关重要的一个对象

       */

      val sc = new SparkContext(conf) //创建SparkContext对象,通过传入SparkConf实例来定制Spark运行的具体参数和配置信息

      

      /**

       * 3步:根据具体的数据来源(HDFSHBaseLocal FSDBS3等)通过SparkContext来创建RDD

       * RDD的创建基本有三种方式:根据外部的数据来源(例如HDFS)、根据Scala集合、由其它的RDD操作

       * 数据会被RDD划分成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴

       */

      //val lines: RDD[String] = sc.textFile("D://Big_Data_Software//spark-1.6.0-bin-hadoop2.6//README.md", 1) //读取本地文件并设置为一个Partion

      val lines = sc.textFile("D://Big_Data_Software//spark-1.6.0-bin-hadoop2.6//README.md", 1) //读取本地文件并设置为一个Partion

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.1步:讲每一行的字符串拆分成单个的单词

       */

      

      val words = lines.flatMap { line => line.split(" ")} //对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合

      

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.2步:在单词拆分的基础上对每个单词实例计数为1,也就是word => (word, 1)

       */

      val pairs = words.map { word => (word, 1) }

      

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.3步:在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数

       */

      val wordCounts = pairs.reduceByKey(_+_) //对相同的Key,进行Value的累计(包括LocalReducer级别同时Reduce

      

      wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))

      

      sc.stop()

      

    }

}

 代码去右击 选择点击 run”wordCount” 来运行程序。生成环境下肯定是写自动化shell脚本自动提交程序的

4. 编写Cluster 模式代码

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.rdd.RDD

/**

 * 使用Scala开发集群运行的Spark WordCount程序

 * @author DT大数据梦工厂

 * 新浪微博:http://weibo.com/ilovepains/

 */

object WordCount_Cluster {

    def main(args: Array[String]){

      /**

       * 1步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,

       * 例如说通过setMaster来设置程序要链接的Spark集群的MasterURL,如果设置

       * local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如

       * 只有1G的内存)的初学者       *

       */

      val conf = new SparkConf() //创建SparkConf对象

      conf.setAppName("Wow,My First Spark App!") //设置应用程序的名称,在程序运行的监控界面可以看到名称

//      conf.setMaster("spark://Master:7077") //此时,程序在Spark集群

      

      /**

       * 2步:创建SparkContext对象

       * SparkContextSpark程序所有功能的唯一入口,无论是采用ScalaJavaPythonR等都必须有一个SparkContext

       * SparkContext核心作用:初始化Spark应用程序运行所需要的核心组件,包括DAGSchedulerTaskSchedulerSchedulerBackend

       * 同时还会负责Spark程序往Master注册程序等

       * SparkContext是整个Spark应用程序中最为至关重要的一个对象

       */

      val sc = new SparkContext(conf) //创建SparkContext对象,通过传入SparkConf实例来定制Spark运行的具体参数和配置信息

      

      /**

       * 3步:根据具体的数据来源(HDFSHBaseLocal FSDBS3等)通过SparkContext来创建RDD

       * RDD的创建基本有三种方式:根据外部的数据来源(例如HDFS)、根据Scala集合、由其它的RDD操作

       * 数据会被RDD划分成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴

       */

      

      //val lines = sc.textFile("hdfs://Master:9000/library/wordcount/input/Data") //读取HDFS文件并切分成不同的Partions

      val lines = sc.textFile("/library/wordcount/input/Data") //读取HDFS文件并切分成不同的Partions

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.1步:讲每一行的字符串拆分成单个的单词

       */

      

      val words = lines.flatMap { line => line.split(" ")} //对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合

      

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.2步:在单词拆分的基础上对每个单词实例计数为1,也就是word => (word, 1)

       */

      val pairs = words.map { word => (word, 1) }

      

      /**

       * 4步:对初始的RDD进行Transformation级别的处理,例如mapfilter等高阶函数等的编程,来进行具体的数据计算

       * 4.3步:在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数

       */

     val wordCountsorderd = pairs.reduceByKey(_+_).map(pair=>(pair._2,pair._1)).sortByKey(false).map(pair =>(pair._1.pair._2)) //对相同的Key,进行Value的累计(包括Local和Reducer级别同时Reduce)

wordCountsorderd.collect.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))

 

      

      sc.stop()

      

    }

}

 

程序达成jar 包

点击 file-> project structure, 在弹出的页面点击 Artifacts,点击右侧的+”,选择jar –> from modules with dependencies,弹出的页面中,设置main class 然后点击ok在弹出页面修改Name(系统生成的name不规范)导出位置删除 scalasparkjar(因为集群环境中已经存在)点击ok 。然后菜单栏中点击build –> Artifacts ,在弹出按钮中,点击bulid,会自动开始打包。

 

在hadoop中执行wordcount方法

将jar 放到linux 系统某个目录中。执行

 ./spark-submit

--class  com.dt.spark. WordCount_Cluster

--master spark://master:7077

/root/documents/sparkapps/wordcount.jar

 

注意事项:

为什么不能再ide开发环境中,直接发布spark程序到spark集群

1. 开发机器的内存和cores的限制,默认情况情况下,spark程序的dirver提交spark程序的机器上, 如果idea中提交程序的话,那idea机器就必须非常强大

2. Dirver指挥workers的运行并频繁的发生同学,如果开发环境和spark集群不在同样一个网络下,就会出现任务丢失,运行缓慢等多种必要的问题。

3. 这是不安全的。

 

 

 

没有更多推荐了,返回首页