Spark-2.WordCount的4种写法

0. 前言

搭好的Spark当然要先写一个最简单的WordCount练练手。 那么,需求是: 1、统计Spark下README.md文件的词频; 2、输出较多,筛选出现次数超过10次的,词频逆序显示

注意

  1. 一般用hdfs上的文件,这里为了简化,用spark目录中的文件。
  2. 读取hdfs用hdfs协议,读取文件用file协议,不写协议默认用file协议读取当前目录下文件。
  3. 要分布式集群上用file读取文件的话,Worker节点的该路径下都需要有这个文件。
  4. 尝鲜而已,怎么简单怎么来,先读取本地文件好了: 4.1 以本地模式运行,设置setMaster("local"); 4.2 local可以接中括号[],括号中的数字表示执行任务的线程数; 4.3 local[*] 表示CPU有几个核就用几个线程。

0.1 构建工程:

这里可以直接构建scala工程,但是一次准备写scala和java的,因此从基础的maven项目开始构建。

0.1.1 Maven quickstart

0.0.1 GroupId+ArtifactId+项目初始化确认点点点

0.0.2 添加Scala Framework 支持

  1. 右键点击项目名,选择Add Framework,选择Scala

image.png

  1. 我已经写过scala项目了,这里默认库选好了。初始化可能不是这个版本,需要创建,点击Use library最右边的Create:

image.png

  1. 确认下拉框中的版本,和Compiler classpath,Standard library中依赖包的版本一致,如果不一致,需要先去pom文件中添加对应版本的依赖,把包虾虾来之后点击这里下面的+号进行依赖的选择 百度有说去Project Structure 的Platform Setting 的Global Library里添加,直接下载。放弃吧,你下不下来的,你下不下来的,你下不下来的。
  2. 我一开始只在pom写了spark的核心依赖,看着maven下包的时候发现一闪而过的Hadoop的默认依赖下的2.6的,又去pom中手动指定了一下。

1. Scala

构建scala工程

package com.junzerg

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

object WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 获取编程入口
      */
    val conf: SparkConf = new SparkConf()
    conf.setAppName("WordCount")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    /**
      * 通过编程入口加载数据
      */
    val linesRdd:RDD[String] = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md")

    /**
      * 对数据进行处理
      */
    val wordCountRdd = linesRdd
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .filter(_._2 > 10)
      .filter(_._1 != "")
      .sortBy(_._2, false)

    /**
      * 对结果数据进行处理l
      */
    wordCountRdd.foreach(println(_))

    sc.stop()
  }
}

1.1 遇到的问题

首次运行报错,百度又是一通鬼扯:

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 10582

打断点,看异常栈,发现最后抛出的异常在:

declaringClass = "com.thoughtworks.paranamer.BytecodeReadingParanamer$ClassReader"
methodName = "accept"
fileName = "BytecodeReadingParanamer.java"
lineNumber = 563

这个类默认下载了2.7版本。和jdk8兼容不是很好,要去pom指定一下:

        <dependency>
            <groupId>com.thoughtworks.paranamer</groupId>
            <artifactId>paranamer</artifactId>
            <version>2.8</version>
        </dependency>

2. Java7(无lambda表达式java)

在IDEA和pom中,退一下版本

/**
 * @author pengjunzhe
 */
public class WordCountJdk7 {
    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) {

        /**
         * 获取编程入口
         */
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("WordCountJdk7");
        JavaSparkContext sc = new JavaSparkContext(conf);

        /**
         * 通过编程入口加载数据
         */
        JavaRDD<String> lines = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md", 2);

        /**
         * 对数据进行处理
         */
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) {
                return new Tuple2<>(s, 1);
            }
        });

        ones = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }).filter(new Function<Tuple2<String, Integer>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Integer> v1) throws Exception {
                return v1._2 > 10;
            }
        }).filter(new Function<Tuple2<String, Integer>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Integer> v1) throws Exception {
                return v1._1.length() > 1;
            }
        });

        JavaPairRDD<Integer, String> transdOnes = ones.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> t2) throws Exception {
                return new Tuple2<>(t2._2, t2._1);
            }
        }).sortByKey();

        ones = transdOnes.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> t2) throws Exception {
                return new Tuple2<>(t2._2, t2._1);
            }
        });


        /**
         * 对结果数据进行处理
         */
        ones.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> t2) throws Exception {
                System.out.println(t2);
            }
        });

        /**
         * 结束关闭入口
         */
        sc.stop();
    }
}

3. Java8(带lambda表达式java)

/**
 * @author pengjunzhe
 */
public class WordCountJdk8 {
    public static void main(String[] args) {

        /**
         * 获取编程入口
         */
        SparkConf conf = new SparkConf();
        conf.setMaster("local").setAppName("WordCountJdk8");
        JavaSparkContext sc = new JavaSparkContext(conf);

        /**
         * 通过编程入口加载数据
         */
        JavaRDD<String> lines = sc.textFile("/Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md", 2);

        /**
         * 对数据进行处理
         */
        JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());

        JavaPairRDD<String, Integer> ones = words
                .mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1))
                .reduceByKey((v1, v2) -> v1 + v2)
                .filter(v1 -> v1._2 > 10)
                .filter(v1 -> v1._1.length() > 1);

        JavaPairRDD<Integer, String> transdOnes = ones
                .mapToPair(t2 -> new Tuple2<>(t2._2, t2._1))
                .sortByKey();

        ones = transdOnes.mapToPair(t2 -> new Tuple2<>(t2._2, t2._1));

        /**
         * 对结果数据进行处理
         */
        // ones.foreach(System.out::println);
        ones.foreach(s -> System.out.println(s));

        /**
         * 结束关闭入口
         */
        sc.stop();
    }
}

注意最后不能用System.out::println的写法,抛出PrintStream不能被序列化的异常。

4.Python

from pyspark import SparkConf, SparkContext

# 获取编程入口
conf = SparkConf() \
    .setMaster("local[*]") \
    .setAppName("WordCount")

sc = SparkContext(conf=conf)

# 通过编程入口加载数据
linesRdd = sc.textFile("file:///Users/pengjunzhe/Downloads/spark-2.4.0-bin-hadoop2.7/README.md")

# 对数据进行处理
wordCountRdd = linesRdd \
    .flatMap(lambda x: x.split(" ")) \
    .map(lambda x: (x, 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .filter(lambda x: x[1] > 10) \
    .filter(lambda x: x[0] != "" ) \
    .sortBy(lambda x: x[1], False)

# 对结果数据进行处理
wordCountRdd.foreach(lambda x: print(x))

# 关闭编程
sc.stop()

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券