Введение в Spark (3) — классическая статистика слов Spark

Большие данные

Искра классическая статистика слов

Подготовить данные

Поскольку мы хотим подсчитывать слова, нам нужен текст, который содержит определенное количество текста. Здесь мы выбираем текст оригинальной английской книги «GoneWithTheWind» («Унесенные ветром»), чтобы сделать статистику данных, чтобы увидеть, как часто каждое слово появляется в статье. Для того, чтобы облегчить всем, чтобы скачать текст. может прибытьGitHubЗагрузите текст и соответствующий код выше. Я помещаю текст в каталог проекта.

Прежде всего, если мы хотим прочитать файл, нам нужно использовать метод textFile в SparkContext, мы пытаемся сначала прочитать первую строку.

Скала реализация

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

object WordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")

    val sc = new SparkContext(conf)

    println(sc.textFile("./GoneWithTheWind").first())
  }

}

Java-реализация

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;

public class WordCountJava {

    public static void main(String[] args){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountJava");

        JavaSparkContext sc = new JavaSparkContext(conf);

        System.out.println(sc.textFile("./GoneWithTheWind").first());
        
    }

}

реализация питона

from pyspark import SparkConf,SparkContext


conf = SparkConf().setMaster("local").setAppName("HelloWorld")

sc = SparkContext(conf=conf)

print(sc.textFile("./GoneWithTheWind").first())

получить вывод

Chapter 1

Возьмем для примера scala, два других языка похожи. На первом этапе мы создали SparkConf

val conf = new SparkConf().setMaster("local").setAppName("WordCount")

Здесь мы устанавливаем для Master значение local, а имя программы — WordCount.Конечно, имя программы можно выбрать произвольно, и неважно, отличается ли оно от имени класса. А вот этот Master не заскринить.Когда запускаем на кластере и используем spark-submit,надо обращать внимание. Сейчас мы обсуждаем только местное письмо, поэтому здесь написано только местное.

Затем мы создаем SparkContext, который является ядром spark, и передаем конфигурацию conf для инициализации.

 val sc = new SparkContext(conf)

Наконец, мы сообщаем SparkContext путь к тексту, а затем выводим первую строку содержимого.

println(sc.textFile("./GoneWithTheWind").first())

начать считать

Затем мы можем начать подсчитывать количество слов в тексте, потому что слова разделены пробелами, поэтому мы можем использовать пробелы в качестве маркера слова.

Скала реализация

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

object WordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")

    val sc = new SparkContext(conf)

    //设置数据路径
    val text = sc.textFile("./GoneWithTheWind")

    //将文本数据按行处理，每行按空格拆成一个数组
    // flatMap会将各个数组中元素合成一个大的集合
    val textSplit = text.flatMap(line =>line.split(" "))

    //处理合并后的集合中的元素，每个元素的值为1，返回一个元组（key,value）
    //其中key为单词，value这里是1，即该单词出现一次
    val textSplitFlag = textSplit.map(word => (word,1))

    //reduceByKey会将textSplitFlag中的key相同的放在一起处理
    //传入的（x,y）中，x是上一次统计后的value，y是本次单词中的value，即每一次是x+1
    val countWord = textSplitFlag.reduceByKey((x,y)=>x+y)

    //将计算后的结果存在项目目录下的result目录中
    countWord.saveAsTextFile("./result")
  }

}

Java-реализация

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 org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class WordCountJava {

    public static void main(String[] args){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountJava");

        JavaSparkContext sc = new JavaSparkContext(conf);

        //设置数据的路径
        JavaRDD<String> textRDD = sc.textFile("./GoneWithTheWind");


        //将文本数据按行处理，每行按空格拆成一个数组,flatMap会将各个数组中元素合成一个大的集合
        //这里需要注意的是FlatMapFunction中<String, String>,第一个表示输入，第二个表示输出
        //与Hadoop中的map-reduce非常相似
        JavaRDD<String> splitRDD = textRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        //处理合并后的集合中的元素，每个元素的值为1，返回一个Tuple2,Tuple2表示两个元素的元组
        //值得注意的是上面是JavaRDD，这里是JavaPairRDD，在返回的是元组时需要注意这个区别
        //PairFunction中<String, String, Integer>，第一个String是输入值类型
        //第二第三个，String, Integer是返回值类型
        //这里返回的是一个word和一个数值1，表示这个单词出现一次
        JavaPairRDD<String, Integer> splitFlagRDD = splitRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s,1);
            }
        });


        //reduceByKey会将splitFlagRDD中的key相同的放在一起处理
        //传入的（x,y）中，x是上一次统计后的value，y是本次单词中的value，即每一次是x+1
        JavaPairRDD<String, Integer> countRDD = splitFlagRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer+integer2;
            }
        });

        //将计算后的结果存在项目目录下的result目录中
        countRDD.saveAsTextFile("./resultJava");



    }

}

реализация питона

from pyspark import SparkConf,SparkContext


conf = SparkConf().setMaster("local").setAppName("HelloWorld")

sc = SparkContext(conf=conf)

# 设置数据的路径
textData = sc.textFile("./GoneWithTheWind")

# 将文本数据按行处理，每行按空格拆成一个数组,flatMap会将各个数组中元素合成一个大的集合
splitData = textData.flatMap(lambda line:line.split(" "))

# 处理合并后的集合中的元素，每个元素的值为1，返回一个元组（key,value）
# 其中key为单词，value这里是1，即该单词出现一次
flagData = splitData.map(lambda word:(word,1))

# reduceByKey会将textSplitFlag中的key相同的放在一起处理
# 传入的（x,y）中，x是上一次统计后的value，y是本次单词中的value，即每一次是x+1
countData = flagData.reduceByKey(lambda x,y:x+y)

#输出文件
countData.saveAsTextFile("./result")

После запуска в живом каталоге получается каталог с именем result.Каталог показан ниже.SUCCESS означает, что файл был успешно сгенерирован, а содержимое файла сохранено в part-00000.

Мы можем просмотреть часть файла:

('Chapter', 1)
('1', 1)
('SCARLETT', 1)
('O’HARA', 1)
('was', 74)
('not', 33)
('beautiful,', 1)
('but', 32)
('men', 4)
('seldom', 3)
('realized', 2)
('it', 37)
('when', 19)
('caught', 1)
('by', 20)
('her', 65)
('charmas', 1)
('the', 336)
('Tarleton', 7)
('twins', 16)
('were.', 1)
('In', 1)
('face', 6)
('were', 49)
...
...
...
...

На этом настоящая программа HelloWorld для искрового подсчета слов завершена. Сравнивая программы на трех языковых версиях, мы обнаружили, что код, написанный на scala и python, очень лаконичен и прост для понимания, а код, реализованный на Java, относительно сложен и труден для понимания. Конечно, это простое для понимания и трудное для понимания понятия являются относительными. Если вы знаете только Java, вы и так должны понимать из нее java-программы, но лаконичные scala и python вам совершенно непонятны. Это не имеет значения, язык — это всего лишь инструмент, все зависит от того, как вы его используете. Более того, мы также можем написать лаконичный код, используя возможности java8.

реализация java8

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 org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class WordCountJava {

    public static void main(String[] args){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountJava");

        JavaSparkContext sc = new JavaSparkContext(conf);

        countJava8(sc);

    }


    public static void countJava8(JavaSparkContext sc){


        sc.textFile("./GoneWithTheWind")
          .flatMap(s->Arrays.asList(s.split(" ")).iterator())
          .mapToPair(s->new Tuple2<>(s,1))
          .reduceByKey((x,y)->x+y)
          .saveAsTextFile("./resultJava8");


    }

}

Превосходство spark отражено в этой небольшой программе.Для подсчета количества вхождений каждого слова в книге spark запускается на одном компьютере (читает файлы, создает временные файлы, записывает результаты на жесткий диск), загружает -run- финиш занял всего 2 секунды.

оптимизировать программу

Может ли программа быть проще и эффективнее? Конечно, мы можем использовать функцию countByValue, которая является широко используемым методом подсчета.

Скала реализация


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

object WordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")

    val sc = new SparkContext(conf)
    

    //设置数据路径
    val text = sc.textFile("./GoneWithTheWind")

    //将文本数据按行处理，每行按空格拆成一个数组
    // flatMap会将各个数组中元素合成一个大的集合
    val textSplit = text.flatMap(line =>line.split(" "))
    
    println(textSplit.countByValue())
  }
}

беги и получай результат

Map(Heknew -> 1, &emsp;&emsp;“Ashley -> 1, “Let’s -> 1, anarresting -> 1, of. -> 1, pasture -> 1, war’s -> 1, wall. -> 1, looks -> 2, ain’t -> 7,.......

Java-реализация

public class WordCountJava {

    public static void main(String[] args){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountJava");

        JavaSparkContext sc = new JavaSparkContext(conf);
        countJava(sc);

    }
    
     public static void countJava(JavaSparkContext sc){
        //设置数据的路径
        JavaRDD<String> textRDD = sc.textFile("./GoneWithTheWind");


        //将文本数据按行处理，每行按空格拆成一个数组,flatMap会将各个数组中元素合成一个大的集合
        //这里需要注意的是FlatMapFunction中<String, String>,第一个表示输入，第二个表示输出
        //与Hadoop中的map-reduce非常相似
        JavaRDD<String> splitRDD = textRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        System.out.println(splitRDD.countByValue());

    }

}

беги и получай результат

{Heknew=1, &emsp;&emsp;“Ashley=1, “Let’s=1, anarresting=1, of.=1, pasture=1, war’s=1, wall.=1, looks=2, ain’t=7, Clayton=1, approval.=1, ideas=1,

реализация питона

from pyspark import SparkConf,SparkContext


conf = SparkConf().setMaster("local").setAppName("HelloWorld")

sc = SparkContext(conf=conf)

# 设置数据的路径
textData = sc.textFile("./GoneWithTheWind")

# 将文本数据按行处理，每行按空格拆成一个数组,flatMap会将各个数组中元素合成一个大的集合
splitData = textData.flatMap(lambda line:line.split(" "))

print(splitData.countByValue())

Запускаем и получаем результат:

defaultdict(<class 'int'>, {'Chapter': 1, '1': 1, 'SCARLETT': 1, 'O’HARA': 1, 'was': 74, 'not': 33, 'beautiful,': 1, 'but': 32, 'men': 4,

Превосходство spark отражено в этой небольшой программе.Для подсчета количества вхождений каждого слова в книге spark запускается на одном компьютере (читает файлы, создает временные файлы, записывает результаты на жесткий диск), загружает -run- финиш занял всего 2 секунды. Если вы хотите получить исходный код и содержимое данных, вы можете перейти на мойgithubскачать.

Категории