spark集群配置以及java操作spark小demo

• spark
  • 安裝
  • 配置
  • 使用java來操作spark

spark

安裝

tar -zxvf spark-2.4.0-bin-hadoop2.7.tgz
rm spark-2.4.0-bin-hadoop2.7.tgz
mv spark-2.4.0-bin-hadoop2.7 sparksudo vim /etc/profileexport SPARK_HOME=/usr/local/stormexport PATH=$PATH:$SPARK_HOME/binsource /etc/profile準備 master worker1 worker2 worker3 這四臺機器首先確保你的Hadoop集群能夠正常運行worker1 worker2 worker3為DataNode, master為NameNode
具體配置參照我的博客https://www.cnblogs.com/ye-hcj/p/10192857.html

配置

1. spark-env.sh

   進入spark的conf目錄下，cp spark-env.sh.template spark-env.shsudo vim spark-env.sh輸入如下配置export JAVA_HOME=/usr/local/jdk/jdk-11.0.1export SCALA_HOME=/usr/local/scala/scalaexport HADOOP_HOME=/usr/local/hadoop/hadoop-3.1.1export SPARK_HOME=/usr/local/spark/sparkexport HADOOP_CONF_DIR=/usr/local/hadoop/hadoop-3.1.1/etc/hadoopexport SPARK_MASTER_HOST=masterexport SPARK_WORKER_MEMORY=1gexport SPARK_WORKER_CORES=1

2. slaves

   進入spark的conf目錄下，cp slaves.template slavessudo vim slaves輸入如下配置    master worker1 worker2 worker3

3. 啟動

   在master中運行 sbin/start-all.sh 即可訪問http://master:8080/即可看到spark的ui

使用java來操作spark

寫個小demo,用來分析10萬個數據中男女人數

1. 模擬數據的java代碼

   // 模擬數據
   // 10萬個人當中，統計青年男性和青年女性的比例，看看男女比例是否均衡
   FileOutputStream f = null;
   ThreadLocalRandom random = ThreadLocalRandom.current();
   String str = "";
   int count = 0;
   try {f = new FileOutputStream("C:\\Users\\26401\\Desktop\\data.txt", true); for(;count<100000;count++) {str = count + " " + random.nextInt(18, 28) + " " + (random.nextBoolean()?'M':'F');f.write((str + "\r\n").getBytes());}} catch (Exception e) {e.printStackTrace();
   } finally {try {if(f != null) f.close();} catch (IOException e) {e.printStackTrace();}
   }

2. 依賴

   <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>test</groupId><artifactId>test</artifactId><version>1.0.0</version><name>test</name><description>Test project for spring boot mybatis</description><packaging>jar</packaging><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.encoding>UTF-8</maven.compiler.encoding><java.version>1.8</java.version><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties> <dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>2.4.0</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-jar-plugin</artifactId><configuration><archive><manifest><addClasspath>true</addClasspath><useUniqueVersions>false</useUniqueVersions><classpathPrefix>lib/</classpathPrefix></manifest></archive></configuration></plugin></plugins></build>
   </project>

3. java代碼

   package test;import java.io.Serializable;import org.apache.spark.SparkConf;
   import org.apache.spark.api.java.JavaRDD;
   import org.apache.spark.api.java.JavaSparkContext;
   import org.apache.spark.api.java.function.Function;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;public class App implements Serializable
   {private static final long serialVersionUID = -7114915627898482737L;public static void main(String[] args) throws Exception {Logger logger=LoggerFactory.getLogger(App.class);SparkConf sparkConf = new SparkConf();sparkConf.setMaster("spark://master:7077");sparkConf.set("spark.submit.deployMode", "cluster");sparkConf.setAppName("FirstTest");JavaSparkContext sc = new JavaSparkContext(sparkConf);JavaRDD<String> file = sc.textFile("hdfs://master:9000/data.txt");JavaRDD<String> male = file.filter(new Function<String, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(String s) throws Exception {logger.info(s);return s.contains("M");}});logger.info("**************************************");logger.info(male.count()+""); // 49991logger.info("**************************************");sc.close();// 其他的api請自行查閱，很簡單，不想看，可以自己瞎點}
   }

4. 運行

   1. 將生成的測試數據data.txt上傳至hdfs
   2. 將打包的jar上傳到master機器
   3. 運行 bin/spark-submit --master spark://master:7077 --class test.App test-1.0.0.jar 
   4. 進入spark的ui界面可以清楚的看到打印的消息