Spark入门（十一）之排序

矫情吗；* 2023-02-24 03:58 10阅读 0赞

###  一、Sort ###

计算文本里面的每个单词出现的个数，单词个数逆序（相同个数单词正序）输出结果。

### 二、maven设置 ###

<?xml version="1.0" encoding="UTF-8"?>
     
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
     
      <groupId>com.mk</groupId>
      <artifactId>spark-test</artifactId>
      <version>1.0</version>
     
      <name>spark-test</name>
      <url>http://spark.mk.com</url>
     
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <scala.version>2.11.1</scala.version>
        <spark.version>2.4.4</spark.version>
        <hadoop.version>2.6.0</hadoop.version>
      </properties>
     
      <dependencies>
        
        <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
        </dependency>
     
        
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
     
     
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.11</version>
          <scope>test</scope>
        </dependency>
      </dependencies>
     
      <build>
        <pluginManagement>
          <plugins>
     
            <plugin>
              <artifactId>maven-clean-plugin</artifactId>
              <version>3.1.0</version>
            </plugin>
     
            <plugin>
              <artifactId>maven-resources-plugin</artifactId>
              <version>3.0.2</version>
            </plugin>
            <plugin>
              <artifactId>maven-compiler-plugin</artifactId>
              <version>3.8.0</version>
            </plugin>
            <plugin>
              <artifactId>maven-surefire-plugin</artifactId>
              <version>2.22.1</version>
            </plugin>
            <plugin>
              <artifactId>maven-jar-plugin</artifactId>
              <version>3.0.2</version>
            </plugin>
          </plugins>
        </pluginManagement>
      </build>
    </project>

### 三、编程代码  ###

public class SortApp implements SparkConfInfo{
    
        public static class SortData implements Comparable<SortData>, Serializable {
            private String word;
            private Integer count;
    
            public SortData(String word, Integer count) {
                this.word = word;
                this.count = count;
            }
    
            public String getWord() {
                return word;
            }
    
            public void setWord(String word) {
                this.word = word;
            }
    
            public Integer getCount() {
                return count;
            }
    
            public void setCount(Integer count) {
                this.count = count;
            }
    
            @Override
            public int compareTo(SortData o) {
    
                if (o == null) {
                    return 1;
                }
    
                int diff = o.count - this.count;
                if (diff != 0)
                    return diff;
    
                if(word == o.word)
                    return 0 ;
    
                if(word == null)
                    return -1;
    
                if(o.word == null)
                    return 1;
    
                 return  this.word.compareTo(o.word);
            }
        }
    
        public static void main(String[]args){
    
            String filePath = "F:\\test\\log.txt";
            SparkSession sparkSession = new SortApp().getSparkConf("sort");
            List<String> wordCounts = sparkSession.sparkContext()
                    .textFile(filePath, 4)
                    .toJavaRDD()
                    .flatMap(v -> Arrays.asList(v.split("[(\\s+)(\r?\n),.。'’]")).iterator())
                    .filter(v -> v.matches("[a-zA-Z-]+"))
                    .map(String::toLowerCase)
                    .mapToPair(v -> new Tuple2<>(v, 1))
                    .reduceByKey(Integer::sum)
                    .map(v->new SortData(v._1, v._2))
                    .sortBy(v -> v, true,4).map(v->v.word)
                    .collect();
    
            wordCounts.forEach(v -> System.out.println(v));
            sparkSession.stop();
        }
    }
    
    public interface SparkConfInfo {
    
        default SparkSession getSparkConf(String appName){
            SparkConf sparkConf = new SparkConf();
            if(System.getProperty("os.name").toLowerCase().contains("win")) {
                sparkConf.setMaster("local[4]");
                System.out.println("使用本地模拟是spark");
            }else
            {
                sparkConf.setMaster("spark://hadoop01:7077,hadoop02:7077,hadoop03:7077");
                sparkConf.set("spark.driver.host","192.168.150.1");//本地ip，必须与spark集群能够相互访问，如：同一个局域网
                sparkConf.setJars(new String[] {".\\out\\artifacts\\spark_test\\spark-test.jar"});//项目构建生成的路径
            }
    
            SparkSession session = SparkSession.builder().appName(appName).config(sparkConf).config(sparkConf).getOrCreate();
            return session;
        }
    }

文件内容

Spark Streaming is an extension of the core Spark API that enables scalable,
      high-throughput, fault-tolerant stream processing of live 
         。data streams. Data， can be ，ingested from many sources like Kafka, Flume, Kinesis, or TCP sockets, and can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems,Spark Streaming provides a high-level abstraction called discretized stream or DStream, which represents a continuous stream of data. DStreams can be created either from input data streams from sources such as Kafka, Flume, and Kinesis, or by applying high-level operations on other DStreams. Internally, a DStream is represented as a sequence of RDDs.
    
    This guide shows you how to start writing Spark Streaming programs with DStreams. You can write Spark Streaming programs in Scala, Java or Python (introduced in Spark 1.2), all of which are presented in this guide. You will find tabs throughout this guide that let you choose between code snippets of different languages. databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

输出

spark
    can
    of
    and
    data
    you
    a
    be
    in
    or
    streaming
    dstreams
    from
    guide
    high-level
    stream
    streams
    this
    algorithms
    as
    dstream
    flume
    is
    kafka
    kinesis
    like
    live
    on
    processed
    processing
    programs
    sources
    that
    to
    which
    with
    abstraction
    all
    an
    api
    apply
    applying
    are
    between
    by
    called
    choose
    code
    complex
    continuous
    core
    created
    dashboards
    databases
    different
    discretized
    either
    enables
    expressed
    extension
    fact
    fault-tolerant
    filesystems
    finally
    find
    functions
    graph
    high-throughput
    how
    input
    internally
    introduced
    java
    join
    languages
    learning
    let
    machine
    many
    map
    operations
    other
    out
    presented
    provides
    pushed
    python
    rdds
    reduce
    represented
    represents
    s
    scala
    scalable
    sequence
    shows
    snippets
    sockets
    start
    such
    tabs
    tcp
    the
    throughout
    using
    will
    window
    write
    writing