正文

Java进行spark计算

IP云
IP云V管理员/2 阅读
11/30
小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

首先在Linux环境安装spark：

可以从如下地址下载最新版本的spark： 

https://spark.apache.org/downloads.html

这个下载下来后是个tgz的压缩包，解压后spark环境就安装好了

或者从github上下载： 

#git clone git://github.com/apache/spark.git

安装好后，进入到spark的根目录，就可以通过spark提供的一些脚本命令行来用spark进行计算了，一个例子 

./bin/spark-submit examples/src/main/python/pi.py 10

这个例子是，可以给脚本命令传入python脚本参数来计算，当然也可以传入Java的jar包参数来计算，即如何计算是通过传入python脚本或者Java类来定义的，

python脚本用spark提供的python api编写，入口是main函数

java类用spark提供的java api来编写，入口也是main函数，要将java类及其依赖的jar包打成jar

下面主要说下在IntelliJ idea中用Java在本地进行spark计算(单线程)

首先需要在pom文件中引入如下两个依赖： 

<dependency>

      <groupId>org.apache.spark</groupId>

      <artifactId>spark-core_2.11</artifactId>

      <version>2.2.0</version>

    </dependency>  

    <dependency>

      <groupId>org.apache.spark</groupId>

      <artifactId>spark-sql_2.11</artifactId>

      <version>2.2.0</version>

</dependency>

Java测试类如下：

功能：分别计算testfile.txt文件中包含字符a和b的行数 

package com.sogou.baike;  

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.SparkSession;  

/**

 * Hello world!

 *

 */

public class SimpleApp {

    public static void main( String[] args ) {

        System.setProperty("hadoop.home.dir", "D:\\software\\hadoop-home\\");

        String logFile = "D:\\testfile.txt";

        SparkSession spark = SparkSession.builder().master("local").appName("Simple Application").getOrCreate();

        Dataset<String> logData = spark.read().textFile(logFile).cache();

        long numAs = logData.filter(s->s.contains("a")).count();

        long numBs = logData.filter(s->s.contains("b")).count();

        System.out.println("a: " + numAs + ", b: " + numBs);

        spark.stop();

    }

}

上面是我已经完全配置好的代码，我最开始运行的过程中遇到了四个错误

第一个错误：

idea不支持java8的新特性lambda表达式

解决办法：参考我的另外一篇博客： 

http://blog.csdn.net/u011734144/article/details/77867697

第二个错误： 

Could not locate executable null\bin\winutils.exe in the Hadoop binaries

解决办法：

到如下地址下载winutils.exe文件， http://public-repo-1.hortonworks.com/hdp-win-alpha/winutils.exe

并保存到如下路径(自行保存路径)： 

D:\\software\\hadoop-home\\

然后在代码中加上这行： 

System.setProperty("hadoop.home.dir", "D:\\software\\hadoop-home\\");

这是配置winutils.exe文件的路径，但是要把winutils.exe文件放到上述目录下的bin/目录中

第三个错误：（这个错误是没有配置spark的集群地址） 

org.apache.spark.SparkException: A master URL must be set in your configuration

解决办法：

可以通过修改idea的配置来解决，实际就是添加jvm的参数配置：

点击idea的Edit Configuration, 在VM options中输入 

-Dspark.master=local

也可以通过修改这行代码： 

SparkSession spark = SparkSession.builder().master("local").appName("Simple Application").getOrCreate();

上面是修改后的，即加上master("local") 就不会报这个错误了，这个表示spark计算访问的是本地spark，而不是远程spark集群

第四个错误： 

Exception in thread "main" java.lang.IllegalArgumentException: System memory 259522560 must be at least 471859200. Please increase heap size using the --driver-memory option or spark.driver.memory in Spark configuration.

    at org.apache.spark.memory.UnifiedMemoryManager$.getMaxMemory(UnifiedMemoryManager.scala:217)

    at org.apache.spark.memory.UnifiedMemoryManager$.apply(UnifiedMemoryManager.scala:199)

    at org.apache.spark.SparkEnv$.create(SparkEnv.scala:332)

    at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:175)

    at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:257)

    at org.apache.spark.SparkContext.<init>(SparkContext.scala:432)

    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2509)

    at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:909)

    at org.apache.spark.sql.SparkSession$Builder$$anonfun$6.apply(SparkSession.scala:901)

    at scala.Option.getOrElse(Option.scala:121)

    at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:901)

    at com.sogou.baike.SimpleApp.main(SimpleApp.java:14)

    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

    at java.lang.reflect.Method.invoke(Method.java:497)

    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

这个说的是项目需要的堆空间太小，所以需要配置java 虚拟机的参数

解决办法：如上，在VM options中添加配置初始堆和最大堆的大小： 

-Xms1024m -Xmx1024m

所以最终的java虚拟机的参数内容为： 

-Dspark.master=local  -Xms1024m -Xmx1024m

上面这种方式的spark计算是在本地执行的，是单线程的

我们其实可以将上述java代码打包成jar包，然后用上面安装好的spark的命令./bin/spark-shell 来执行spark计算，并且可以用该命令将计算提交到集群上计算，加快速度

注意spark命令要运行必须配置JAVA_HOME为java8

相关阅读

目录[+]