Spark-2.WordCount的4种写法

悠扬前奏

发布于 2019-05-30 15:01:28

1.5K0

发布于 2019-05-30 15:01:28

文章被收录于专栏：悠扬前奏的博客

0. 前言

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

注意：

一般用hdfs上的文件，这里为了简化，用spark目录中的文件。
读取hdfs用hdfs协议，读取文件用file协议，不写协议默认用file协议读取当前目录下文件。
要分布式集群上用file读取文件的话，Worker节点的该路径下都需要有这个文件。
尝鲜而已，怎么简单怎么来，先读取本地文件好了： 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 支持

右键点击项目名，选择Add Framework，选择Scala

image.png

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

image.png

确认下拉框中的版本，和Compiler classpath，Standard library中依赖包的版本一致，如果不一致，需要先去pom文件中添加对应版本的依赖，把包虾虾来之后点击这里下面的+号进行依赖的选择 百度有说去Project Structure 的Platform Setting 的Global Library里添加，直接下载。放弃吧，你下不下来的，你下不下来的，你下不下来的。
我一开始只在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()

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2019.03.13 ，如有侵权请联系 cloudcommunity@tencent.com 删除

scala