前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Spark-2.WordCount的4种写法

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次的,词频逆序显示

注意

  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工程

代码语言:javascript
复制
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 遇到的问题

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

代码语言:javascript
复制
Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 10582

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

代码语言:javascript
复制
declaringClass = "com.thoughtworks.paranamer.BytecodeReadingParanamer$ClassReader"
methodName = "accept"
fileName = "BytecodeReadingParanamer.java"
lineNumber = 563

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

代码语言:javascript
复制
        <dependency>
            <groupId>com.thoughtworks.paranamer</groupId>
            <artifactId>paranamer</artifactId>
            <version>2.8</version>
        </dependency>

2. Java7(无lambda表达式java)

在IDEA和pom中,退一下版本

代码语言:javascript
复制
/**
 * @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)

代码语言:javascript
复制
/**
 * @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

代码语言:javascript
复制
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 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 0. 前言
    • 0.1 构建工程:
      • 0.1.1 Maven quickstart
      • 0.0.1 GroupId+ArtifactId+项目初始化确认点点点
      • 0.0.2 添加Scala Framework 支持
  • 1. Scala
    • 1.1 遇到的问题
    • 2. Java7(无lambda表达式java)
    • 3. Java8(带lambda表达式java)
    • 4.Python
    相关产品与服务
    大数据
    全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
    领券
    问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档