Spark入门系列(2)-Spark开发WorldCount入门案例

梦里梦外; 2023-10-10 12:27 15阅读 0赞

### 环境准备 ###

1.  安装并配置Scala开发环境  
    [文档][Link 1]
2.  创建一个普通maven父项目，并创建子模块。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70]
3.  添加scala库  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 1]
4.  编写一个测试程序，检验环境  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 2]
5.  添加pom

<dependencies>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
        </dependencies>
        <build>
            <plugins>
                
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                    <executions>
                        <execution>
                            
                            <goals>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <version>3.1.0</version>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                    </configuration>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>

1.  resources目录下添加日志文件log4j.properties

log4j.rootCategory=ERROR, console
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.target=System.err
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern=%d{
        yy/MM/dd 
    HH:mm:ss} %p %c{
        1}: %m%n
    # Set the default spark-shell log level to ERROR. When running the spark-shell,
    the
    # log level for this class is used to overwrite the root logger's log level, so
    that
    # the user can have different defaults for the shell and regular Spark apps.
    log4j.logger.org.apache.spark.repl.Main=ERROR
    # Settings to quiet third party logs that are too verbose
    log4j.logger.org.spark_project.jetty=ERROR
    log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
    log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=ERROR
    log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=ERROR
    log4j.logger.org.apache.parquet=ERROR
    log4j.logger.parquet=ERROR
    # SPARK-9183: Settings to avoid annoying messages when looking up nonexistent
    UDFs in SparkSQL with Hive support
    log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
    log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR

### 代码实现 ###

1.  编写WorldCount

object WorldCount {
        
      def main(args: Array[String]): Unit = {
        
        // 建立连接
        val sparkConf = new SparkConf().setMaster("local").setAppName("WorldCount")
        // 创建 Spark 上下文环境对象（连接对象）
        val sc = new SparkContext(sparkConf)
        // 读取文件数据
        val fileRDD: RDD[String] = sc.textFile("D:\\input")
        // 将文件中的数据进行分词
        val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
        // 转换数据结构 word => (word, 1)
        val word2OneRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
        // 将转换结构后的数据按照相同的单词进行分组聚合
        val word2CountRDD: RDD[(String, Int)] = word2OneRDD.reduceByKey(_ + _)
        // 将数据聚合结果采集到内存中
        val word2Count: Array[(String, Int)] = word2CountRDD.collect()
        // 打印结果
        word2Count.foreach(println)
        sc.stop()
      }
    }

1.  D:\\input目录下创建测试文件并运行程序

"C:\Program Files\Java\jdk1.8.0_151\bin\java.exe" "-org.pearl.spark.core.demo.worldcount.WorldCount
    (asadasdavsadasda,356)
    (das,5518)
    (avvsadasda,178)
    (a,1248)
    (asadasda,2134)
    (sadasda,1960)
    (asdasd,5518)
    (av,178)
    (asadasdasadasda,1068)
    (asadasdav,356)
    
    Process finished with exit code 0

[Link 1]: https://blog.csdn.net/qq_43437874/article/details/110919808
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210126160801252.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210126161346492.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210126161454200.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDM3ODc0,size_16,color_FFFFFF,t_70