大数据-86 Spark 集群 WordCount 用 Scala & Java 调用Spark 编译并打包上传运行 梦开始的地方

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 大数据-86 Spark 集群 WordCount 用 Scala & Java 调用Spark 编译并打包上传运行 梦开始的地方

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka(已更完)

Spark(正在更新!)

章节内容

上节我们完成了如下的内容:


Spark RDD 操作方式Action

Spark RDD的 Key-Value RDD

详细解释与测试案例

梦的开始

写一个WordCount程序虽然看似简单,但它在大数据学习中有着深远的意义。就像编程世界中的“Hello World”,WordCount是我们迈入分布式计算世界的第一步。在这个过程中,我不仅加深了对Spark生态系统的理解,还亲身体验了大数据处理的核心思想:分而治之。


通过编写和运行这个程序,我意识到,尽管代码本身很简单,但其背后的概念却揭示了大数据处理的复杂性与挑战性。每个词频的统计背后,都代表着分布式系统中对数据的高效切分、分发和聚合。这使我更加意识到,在大数据的世界里,性能优化和资源管理是永恒的主题。


更重要的是,WordCount让我感受到Scala语言在处理并行计算时的优势。通过在实际环境中部署和运行这个程序,我也看到了自己从理论学习向实践应用迈出的重要一步。这不仅是一段代码的完成,更是我在大数据领域探索旅程的一个重要里程碑。


总的来说,这段经历让我更加坚定了继续深入学习和应用大数据技术的决心。WordCount不仅是学习的起点,更是打开大数据世界大门的一把钥匙。


环境依赖

首先要确保你之前的环境都搭建完毕了,最起码的要有单机的Spark,最好是有Spark集群,可以更好的进行学习和测试。


导入依赖

<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>icu.wzk</groupId>
    <artifactId>spark-wordcount</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <scala.version>2.12.10</scala.version>
        <spark.version>2.4.5</spark.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>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <dependency>
            <groupId>com.typesafe</groupId>
            <artifactId>config</artifactId>
            <version>1.3.4</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>4.4.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>cn.lagou.sparkcore.WordCount</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

编写Scala

使用Scala完成我们的Word Count程序:

package icu.wzk

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

object WordCount {
  def main(args: Array[String]): Unit = {
    var conf = new SparkConf().setAppName("ScalaHelloWorldCount")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val lines: RDD[String] = sc.textFile(args(0))

    val words: RDD[String] = lines.flatMap(line => line.split("\\s+"))
    val wordMap: RDD[(String, Int)] = words.map(x => (x, 1))
    val result: RDD[(String, Int)] = wordMap.reduceByKey(_ + _)

    result.foreach(println)
    sc.stop()
  }
}

大致的项目结构和内容,如下图所示:

编译项目

运行Maven的Package,等待执行完毕后,会在 target 下打包出一个 Jar 包。

如果是第一次打包,需要下载包,时间会比较久。

# 你也可以用Shell的方式
mvn clean package

运行的过程如下图所示:

打包完的结果大致如下:

上传项目

将项目上传到Spark的集群中:

cd /opt/wzk

我上传到该目录,该目录的情况大致如下:

运行项目

编写如下的指令,将任务提交到Spark集群中进行运行。

我这里随便找了个文件,你也可以找个文件进行运行。

spark-submit --master local[*] --class icu.wzk.WordCount spark-wordcount-1.0-SNAPSHOT.jar /opt/wzk/goodtbl.java

运行结果如下图:

经过一段时间的计算之后,可以看到最终的结果如下图所示:

编写Java

package icu.wzk;

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 scala.Tuple2;

import java.util.Arrays;

public class JavaWordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("JavaWordCount")
                .setMaster("local[*]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("WARN");
        JavaRDD<String> lines = sc.textFile(args[0]);
        JavaRDD<String> words = lines
                .flatMap(line -> Arrays.stream(line.split("\\s+")).iterator());
        JavaPairRDD<String, Integer> wordsMap = words
                .mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> results = wordsMap.reduceByKey((x, y) -> x + y);
        results.foreach(elem -> System.out.println(elem));
        sc.stop();
    }
}

编译项目

和上面一样,Scala的方式一样:

上传项目

同样的,和上述的Scala的过程一样,将项目上传:

/opt/wzk/spark-wordcount-1.0-SNAPSHOT.jar

运行项目

这里注意,写的是Java的类,而不是Scala的启动:

spark-submit --master local[*] --class icu.wzk.JavaWordCount spark-wordcount-1.0-SNAP

运行的过程截图如下图所示:

等待执行完毕,最终的结果如下图所示:

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
2月前
|
分布式计算 大数据 Java
大数据-87 Spark 集群 案例学习 Spark Scala 案例 手写计算圆周率、计算共同好友
大数据-87 Spark 集群 案例学习 Spark Scala 案例 手写计算圆周率、计算共同好友
59 5
|
2月前
|
分布式计算 关系型数据库 MySQL
大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL
大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL
52 3
|
2月前
|
IDE Java 编译器
Java:如何确定编译和运行时类路径是否一致
类路径(Classpath)是JVM用于查找类文件的路径列表,对编译和运行Java程序至关重要。编译时通过`javac -classpath`指定,运行时通过`java -classpath`指定。IDE如Eclipse和IntelliJ IDEA也提供界面管理类路径。确保编译和运行时类路径一致,特别是外部库和项目内部类的路径设置。
148 5
|
2月前
|
小程序 Oracle Java
JVM知识体系学习一:JVM了解基础、java编译后class文件的类结构详解,class分析工具 javap 和 jclasslib 的使用
这篇文章是关于JVM基础知识的介绍,包括JVM的跨平台和跨语言特性、Class文件格式的详细解析,以及如何使用javap和jclasslib工具来分析Class文件。
56 0
JVM知识体系学习一:JVM了解基础、java编译后class文件的类结构详解,class分析工具 javap 和 jclasslib 的使用
|
3月前
|
消息中间件 分布式计算 Java
Linux环境下 java程序提交spark任务到Yarn报错
Linux环境下 java程序提交spark任务到Yarn报错
47 5
|
2月前
|
SQL 分布式计算 Java
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
50 0
|
2月前
|
缓存 分布式计算 大数据
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(一)
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(一)
55 0
|
2月前
|
分布式计算 算法 大数据
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(二)
大数据-90 Spark 集群 RDD 编程-高阶 RDD容错机制、RDD的分区、自定义分区器(Scala编写)、RDD创建方式(二)
56 0
|
Java Maven 数据安全/隐私保护
【jar】JDK将单个的java文件打包为jar包,并引用到项目中使用【MD5加密】
========================================================================================================================= 后文心得附录:【实际操作使用过程中的使用心得】   1.
1453 0
|
Java 中间件
【JAVA系列】打包Class文件生成Jar包
【JAVA系列】打包Class文件生成Jar包
139 0