一、Spark介紹

Apache Spark is a fast and general-purpose cluster computing system. It provides high-level APIs in Java, Scala, Python and R, and an optimized engine that supports general execution graphs. It also supports a rich set of higher-level tools including Spark SQL for SQL and structured data processing, MLlib for machine learning, GraphX for graph processing, and Spark Streaming.

Spark是一個快速且多功能的集群計算系統。它為多種不同語言提供高級API，和支持一般執行圖的優化引擎。它也有豐富的高級工具集，Spark SQL進行結構化數據的處理，MLib處理機器學習，GraphX進行圖處理，以及Spark Streaming流計算。

組成

它的主要組件有：

• SparkCore
  • 將分佈式數據抽象為彈性分佈式數據集（RDD），實現了應用任務調度、RPC、序列化和壓縮，併為運行在其上的上層組件提供API。
• SparkSQL
  • Spark Sql 是Spark來操作結構化數據的程序包，可以讓我使用SQL語句的方式來查詢數據，Spark支持 多種數據源，包含Hive表，parquest以及JSON等內容。
• SparkStreaming
  • 是Spark提供的實時數據進行流式計算的組件。
• MLlib
  • 提供常用機器學習算法的實現庫。
• GraphX
  • 提供一個分佈式圖計算框架，能高效進行圖計算。
• BlinkDB
  • 用於在海量數據上進行交互式SQL的近似查詢引擎。
• Tachyon
  • 以內存為中心高容錯的的分佈式文件系統。

返回一個包含數據集前n個元素的數組

二、WordCount程序講解

編寫代碼

scala程序編寫

object WordCountDemo {

  def main(args: Array[String]): Unit = {
    //創建Spark配置對象
    val conf = new SparkConf().setMaster("local").setAppName("MyApp")
    //通過conf創建sc
    val sc = new SparkContext(conf)
    //讀取文件
    val rdd1 = sc.textFile("/Users/README.md")
    //計算
    val rdd2 = rdd1.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey(_+_)
    //打印
    rdd2.take(10).foreach(println)

  }

}

java程序編寫

public class WordCountJavaDemo {

    public static void main(String[] args) {
        
        SparkConf conf = new SparkConf();
        conf.setAppName("myapp").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> rdd1 = sc.textFile("/Users/README.md");
        JavaRDD<String> rdd2 = rdd1.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                List<String> list = new ArrayList<>();
                String[] arr = s.split(" ");
                for (String ss : arr) {
                    list.add(ss);
                }
                return list.iterator();
            }
        });

        JavaPairRDD<String, Integer> rdd3 = rdd2.mapToPair(new PairFunction<String, String, Integer>() {

            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }

        });

        JavaPairRDD<String, Integer> rdd4 = rdd3.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        List<Tuple2<String, Integer>> list = rdd4.collect();
        for (Tuple2<String, Integer> t : list) {
            System.out.println(t._1() + " " + t._2());
        }
    }

}

三、原理介紹

RDD

• 由一系列Partition組成
• RDD之間有一系列依賴關係
• RDD每個算子實際上是作用在每個Partition上
• RDD會提供一系列最佳位置
• 分區器是作用在KV格式的RDD上

RDD會在多個節點上存儲，就和hdfs的分佈式道理是一樣的。hdfs文件被切分為多個block存儲在各個節點上，而RDD是被切分為多個partition。不同的partition可能在不同的節點上。

Spark執行流程

1、Driver
分發task，在分發之前，會調用RDD的方法，獲取partition的位置。
將task的計算結果，拉回到Driver端
Driver是一個JVM進程

2、Worker

寬依賴、窄依賴

圖中stage2的并行度是4，也就是有4個task。

寬依賴

父RDD與子RDD，partition的關係是一對多，就是寬依賴。寬依賴於shuffle對應。

窄依賴

父RDD與子RDD，partition的關係是一對一或多對一，就是窄依賴。

四、Spark常用算子

Transformation算子

特點：懶執行

（1）map

map的輸入變換函數應用於RDD中所有元素

（2）flatMap

flatMap與map區別在於map為“映射”，而flatMap“先映射，后扁平化”，map對每一次（func）都產生一個元素，返回一個對象，而flatMap多一步就是將所有對象合併為一個對象。

（3）flatMapValues

每個元素的Value被輸入函數映射為一系列的值，然後這些值再與原RDD中的Key組成一系列新的KV對。

代碼

x = sc.parallelize([("a", ["x", "y", "z"]), ("b", ["p", "r"])])
def f(x): return x
x.flatMapValues(f).collect()

打印結果

 [('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]

filter

過濾操作，滿足filter內function函數為true的RDD內所有元素組成一個新的數據集。

（4）groupByKey

主要作用是將相同的所有的鍵值對分組到一個集合序列當中，其順序是不確定的。

（5）reduceByKey

與groupByKey類似，卻有不同。如(a,1), (a,2), (b,1), (b,2)。groupByKey產生中間結果為( (a,1), (a,2) ), ( (b,1), (b,2) )。而reduceByKey為(a,3), (b,3)。

reduceByKey主要作用是聚合，groupByKey主要作用是分組。

（6）take

Action算子

特點：立即觸發執行

五、SparkSQL

介紹

Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed. Internally, Spark SQL uses this extra information to perform extra optimizations. There are several ways to interact with Spark SQL including SQL and the Dataset API. When computing a result the same execution engine is used, independent of which API/language you are using to express the computation. This unification means that developers can easily switch back and forth between different APIs based on which provides the most natural way to express a given transformation.

SparkSQL是Spark的一個用來處理結構化數據的模塊。使用類似SQL的方式訪問Hadoop，實現MR計算。

Datasets的概念

A Dataset is a distributed collection of data. Dataset is a new interface added in Spark 1.6 that provides the benefits of RDDs (strong typing, ability to use powerful lambda functions) with the benefits of Spark SQL’s optimized execution engine. A Dataset can be constructed from JVM objects and then manipulated using functional transformations (map, flatMap, filter, etc.). The Dataset API is available in Scala and Java. Python does not have the support for the Dataset API. But due to Python’s dynamic nature, many of the benefits of the Dataset API are already available (i.e. you can access the field of a row by name naturally row.columnName). The case for R is similar.

Dataset是分佈式數據集合。

DataFrames概念

A DataFrame is a Dataset organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs. The DataFrame API is available in Scala, Java, Python, and R. In Scala and Java, a DataFrame is represented by a Dataset of Rows. In the Scala API, DataFrame is simply a type alias of Dataset[Row]. While, in Java API, users need to use Dataset to represent a DataFrame.

基本使用

（1）創建DataFrames

數據

{"id":"1","name":"zhangsan","age":"12"}
{"id":"2","name":"lisi","age":"12"}
{"id":"3","name":"wangwu","age":"12"}

代碼

object SparkSqlDemo {

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

    //創建Spark配置對象
    val conf = new SparkConf().setMaster("local[4]").setAppName("MyApp");

    val spark = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .config(conf)
      .getOrCreate()

    val df = spark.read.json("/Users/opensource/dev-problem/source/people_sample_json.json");
    df.show()

  }

}

（2）查詢

val df = spark.read.json("/Users/fangzhijie/opensource/dev-problem/source/people_sample_json.json");
df.createOrReplaceTempView("people")
val sqlDF = spark.sql("SELECT * FROM people WHERE name = 'zhangsan'")
sqlDF.show()

六、SparkStreaming

介紹

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, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.

基本使用

（1）簡單使用

object SparkStreamingDemo {

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

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    //創建Spark流上下文
    val ssc = new StreamingContext(conf, Seconds(1))
    //創建Socket文本流
    val lines = ssc.socketTextStream("localhost", 9999)
    val words = lines.flatMap(_.split(" "))

    val pairs = words.map(word => (word, 1))
    val wordCounts = pairs.reduceByKey(_ + _)

    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()
    //啟動
    ssc.start()
    //等待結束
    ssc.awaitTermination()  // Wait for the computation to terminate


  }

}

使用shell命令監聽端口，輸入待計算內容

$ nc -lk 9999

原理

SparkStreaming的編程抽象是離散化流（DStream），它是一個RDD序列，每個RDD代表數據流中一個時間片內的數據。

參考文檔

《Spark快速大數據分析》

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!!

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!

※大陸寄台灣空運注意事項

※大陸海運台灣交貨時間多久?

※避免吃悶虧無故遭抬價!台中搬家公司免費估價,有契約讓您安心有保障!

本文已經收錄自 (61k+Star!【Java學習+面試指南】 一份涵蓋大部分Java程序員所需要掌握的核心知識。歡迎 Star!歡迎騷擾！)
原文地址：https://javaguide.cn/2019/08/20/java/java%E5%9F%BA%E7%A1%80/Java%E7%96%91%E9%9A%BE%E7%82%B9/

1. 基礎

1.1. 正確使用 equals 方法

Object的equals方法容易拋空指針異常，應使用常量或確定有值的對象來調用 equals。

舉個例子：

// 不能使用一個值為null的引用類型變量來調用非靜態方法，否則會拋出異常
String str = null;
if (str.equals("SnailClimb")) {
  ...
} else {
  ..
}

運行上面的程序會拋出空指針異常，但是我們把第二行的條件判斷語句改為下面這樣的話，就不會拋出空指針異常，else 語句塊得到執行。：

"SnailClimb".equals(str);// false 

不過更推薦使用 java.util.Objects#equals(JDK7 引入的工具類)。

Objects.equals(null,"SnailClimb");// false

我們看一下java.util.Objects#equals的源碼就知道原因了。

public static boolean equals(Object a, Object b) {
        // 可以避免空指針異常。如果a==null的話此時a.equals(b)就不會得到執行，避免出現空指針異常。
        return (a == b) || (a != null && a.equals(b));
    }

注意：

Reference:

• 每種原始類型都有默認值一樣，如int默認值為 0，boolean 的默認值為 false，null 是任何引用類型的默認值，不嚴格的說是所有 Object 類型的默認值。
• 可以使用 == 或者 != 操作來比較null值，但是不能使用其他算法或者邏輯操作。在Java中null == null將返回true。
• 不能使用一個值為null的引用類型變量來調用非靜態方法，否則會拋出異常

1.2. 整型包裝類值的比較

所有整型包裝類對象值的比較必須使用equals方法。

先看下面這個例子：

Integer x = 3;
Integer y = 3;
System.out.println(x == y);// true
Integer a = new Integer(3);
Integer b = new Integer(3);
System.out.println(a == b);//false
System.out.println(a.equals(b));//true

當使用自動裝箱方式創建一個Integer對象時，當數值在-128 ~127時，會將創建的 Integer 對象緩存起來，當下次再出現該數值時，直接從緩存中取出對應的Integer對象。所以上述代碼中，x和y引用的是相同的Integer對象。

注意：如果你的IDE(IDEA/Eclipse)上安裝了阿里巴巴的p3c插件，這個插件如果檢測到你用 ==的話會報錯提示，推薦安裝一個這個插件，很不錯。

1.3. BigDecimal

1.3.1. BigDecimal 的用處

《阿里巴巴Java開發手冊》中提到：浮點數之間的等值判斷，基本數據類型不能用==來比較，包裝數據類型不能用 equals 來判斷。 具體原理和浮點數的編碼方式有關，這裏就不多提了，我們下面直接上實例：

float a = 1.0f - 0.9f;
float b = 0.9f - 0.8f;
System.out.println(a);// 0.100000024
System.out.println(b);// 0.099999964
System.out.println(a == b);// false

具有基本數學知識的我們很清楚的知道輸出並不是我們想要的結果（精度丟失），我們如何解決這個問題呢？一種很常用的方法是：使用使用 BigDecimal 來定義浮點數的值，再進行浮點數的運算操作。

BigDecimal a = new BigDecimal("1.0");
BigDecimal b = new BigDecimal("0.9");
BigDecimal c = new BigDecimal("0.8");
BigDecimal x = a.subtract(b);// 0.1
BigDecimal y = b.subtract(c);// 0.1
System.out.println(x.equals(y));// true 

1.3.2. BigDecimal 的大小比較

a.compareTo(b) : 返回 -1 表示小於，0 表示 等於， 1表示 大於。

BigDecimal a = new BigDecimal("1.0");
BigDecimal b = new BigDecimal("0.9");
System.out.println(a.compareTo(b));// 1

1.3.3. BigDecimal 保留幾位小數

通過 setScale方法設置保留幾位小數以及保留規則。保留規則有挺多種，不需要記，IDEA會提示。

BigDecimal m = new BigDecimal("1.255433");
BigDecimal n = m.setScale(3,BigDecimal.ROUND_HALF_DOWN);
System.out.println(n);// 1.255

1.3.4. BigDecimal 的使用注意事項

注意：我們在使用BigDecimal時，為了防止精度丟失，推薦使用它的 BigDecimal(String) 構造方法來創建對象。《阿里巴巴Java開發手冊》對這部分內容也有提到如下圖所示。

1.3.5. 總結

BigDecimal 主要用來操作（大）浮點數，BigInteger 主要用來操作大整數（超過 long 類型）。

BigDecimal 的實現利用到了 BigInteger, 所不同的是 BigDecimal 加入了小數位的概念

1.4. 基本數據類型與包裝數據類型的使用標準

Reference:《阿里巴巴Java開發手冊》

• 【強制】所有的 POJO 類屬性必須使用包裝數據類型。
• 【強制】RPC 方法的返回值和參數必須使用包裝數據類型。
• 【推薦】所有的局部變量使用基本數據類型。

比如我們如果自定義了一個Student類,其中有一個屬性是成績score,如果用Integer而不用int定義,一次考試,學生可能沒考,值是null,也可能考了,但考了0分,值是0,這兩個表達的狀態明顯不一樣.

說明 :POJO 類屬性沒有初值是提醒使用者在需要使用時，必須自己顯式地進行賦值，任何 NPE 問題，或者入庫檢查，都由使用者來保證。

正例 : 數據庫的查詢結果可能是 null，因為自動拆箱，用基本數據類型接收有 NPE 風險。

反例 : 比如显示成交總額漲跌情況，即正負 x%，x 為基本數據類型，調用的 RPC 服務，調用不成功時，返回的是默認值，頁面显示為 0%，這是不合理的，應該显示成中劃線。所以包裝數據類型的 null 值，能夠表示額外的信息，如:遠程調用失敗，異常退出。

2. 集合

2.1. Arrays.asList()使用指南

最近使用Arrays.asList()遇到了一些坑，然後在網上看到這篇文章： 感覺挺不錯的，但是還不是特別全面。所以，自己對於這塊小知識點進行了簡單的總結。

2.1.1. 簡介

Arrays.asList()在平時開發中還是比較常見的，我們可以使用它將一個數組轉換為一個List集合。

String[] myArray = { "Apple", "Banana", "Orange" }； 
List<String> myList = Arrays.asList(myArray);
//上面兩個語句等價於下面一條語句
List<String> myList = Arrays.asList("Apple","Banana", "Orange");

JDK 源碼對於這個方法的說明：

/**
 *返回由指定數組支持的固定大小的列表。此方法作為基於數組和基於集合的API之間的橋樑，與           Collection.toArray()結合使用。返回的List是可序列化並實現RandomAccess接口。
 */ 
public static <T> List<T> asList(T... a) {
    return new ArrayList<>(a);
}

2.1.2. 《阿里巴巴Java 開發手冊》對其的描述

Arrays.asList()將數組轉換為集合后,底層其實還是數組，《阿里巴巴Java 開發手冊》對於這個方法有如下描述：

2.1.3. 使用時的注意事項總結

傳遞的數組必須是對象數組，而不是基本類型。

Arrays.asList()是泛型方法，傳入的對象必須是對象數組。

int[] myArray = { 1, 2, 3 };
List myList = Arrays.asList(myArray);
System.out.println(myList.size());//1
System.out.println(myList.get(0));//數組地址值
System.out.println(myList.get(1));//報錯：ArrayIndexOutOfBoundsException
int [] array=(int[]) myList.get(0);
System.out.println(array[0]);//1

當傳入一個原生數據類型數組時，Arrays.asList() 的真正得到的參數就不是數組中的元素，而是數組對象本身！此時List 的唯一元素就是這個數組，這也就解釋了上面的代碼。

我們使用包裝類型數組就可以解決這個問題。

Integer[] myArray = { 1, 2, 3 };

使用集合的修改方法:add()、remove()、clear()會拋出異常。

List myList = Arrays.asList(1, 2, 3);
myList.add(4);//運行時報錯：UnsupportedOperationException
myList.remove(1);//運行時報錯：UnsupportedOperationException
myList.clear();//運行時報錯：UnsupportedOperationException

Arrays.asList() 方法返回的並不是 java.util.ArrayList ，而是 java.util.Arrays 的一個內部類,這個內部類並沒有實現集合的修改方法或者說並沒有重寫這些方法。

List myList = Arrays.asList(1, 2, 3);
System.out.println(myList.getClass());//class java.util.Arrays$ArrayList

下圖是java.util.Arrays$ArrayList的簡易源碼，我們可以看到這個類重寫的方法有哪些。

  private static class ArrayList<E> extends AbstractList<E>
        implements RandomAccess, java.io.Serializable
    {
        ...

        @Override
        public E get(int index) {
          ...
        }

        @Override
        public E set(int index, E element) {
          ...
        }

        @Override
        public int indexOf(Object o) {
          ...
        }

        @Override
        public boolean contains(Object o) {
           ...
        }

        @Override
        public void forEach(Consumer<? super E> action) {
          ...
        }

        @Override
        public void replaceAll(UnaryOperator<E> operator) {
          ...
        }

        @Override
        public void sort(Comparator<? super E> c) {
          ...
        }
    }

我們再看一下java.util.AbstractList的remove()方法，這樣我們就明白為啥會拋出UnsupportedOperationException。

public E remove(int index) {
    throw new UnsupportedOperationException();
}

2.1.4. 如何正確的將數組轉換為ArrayList?

stackoverflow：https://dwz.cn/vcBkTiTW

1. 自己動手實現（教育目的）

//JDK1.5+
static <T> List<T> arrayToList(final T[] array) {
  final List<T> l = new ArrayList<T>(array.length);

  for (final T s : array) {
    l.add(s);
  }
  return (l);
}

Integer [] myArray = { 1, 2, 3 };
System.out.println(arrayToList(myArray).getClass());//class java.util.ArrayList

2. 最簡便的方法(推薦)

List list = new ArrayList<>(Arrays.asList("a", "b", "c"))

3. 使用 Java8 的Stream(推薦)

Integer [] myArray = { 1, 2, 3 };
List myList = Arrays.stream(myArray).collect(Collectors.toList());
//基本類型也可以實現轉換（依賴boxed的裝箱操作）
int [] myArray2 = { 1, 2, 3 };
List myList = Arrays.stream(myArray2).boxed().collect(Collectors.toList());

4. 使用 Guava(推薦)

對於不可變集合，你可以使用類及其與工廠方法：（參數不能為空）

List<String> il = ImmutableList.of("string", "elements");  // from varargs
List<String> il = ImmutableList.copyOf(aStringArray);      // from array

對於可變集合，你可以使用類及其工廠方法：

List<String> l1 = Lists.newArrayList(anotherListOrCollection);    // from collection
List<String> l2 = Lists.newArrayList(aStringArray);               // from array
List<String> l3 = Lists.newArrayList("or", "string", "elements"); // from varargs

5. 使用 Apache Commons Collections

List<String> list = new ArrayList<String>();
CollectionUtils.addAll(list, str);

2.2. Collection.toArray()方法使用的坑&如何反轉數組

該方法是一個泛型方法：<T> T[] toArray(T[] a); 如果toArray方法中沒有傳遞任何參數的話返回的是Object類型數組。

String [] s= new String[]{
    "dog", "lazy", "a", "over", "jumps", "fox", "brown", "quick", "A"
};
List<String> list = Arrays.asList(s);
Collections.reverse(list);
s=list.toArray(new String[0]);//沒有指定類型的話會報錯

由於JVM優化，new String[0]作為Collection.toArray()方法的參數現在使用更好，new String[0]就是起一個模板的作用，指定了返回數組的類型，0是為了節省空間，因為它只是為了說明返回的類型。詳見：

2.3. 不要在 foreach 循環里進行元素的 remove/add 操作

如果要進行remove操作，可以調用迭代器的 remove方法而不是集合類的 remove 方法。因為如果列表在任何時間從結構上修改創建迭代器之後，以任何方式除非通過迭代器自身remove/add方法，迭代器都將拋出一個ConcurrentModificationException,這就是單線程狀態下產生的 fail-fast 機制。

fail-fast 機制 ：多個線程對 fail-fast 集合進行修改的時，可能會拋出ConcurrentModificationException，單線程下也會出現這種情況，上面已經提到過。

java.util包下面的所有的集合類都是fail-fast的，而java.util.concurrent包下面的所有的類都是fail-safe的。

開源項目推薦

作者的其他開源項目推薦：

1. ：【Java學習+面試指南】 一份涵蓋大部分Java程序員所需要掌握的核心知識。
2. : 適合新手入門以及有經驗的開發人員查閱的 Spring Boot 教程（業餘時間維護中，歡迎一起維護）。
3. : 我覺得技術人員應該有的一些好習慣！
4. :從零入門 ！Spring Security With JWT（含權限驗證）後端部分代碼。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!!

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!

※大陸寄台灣空運注意事項

※大陸海運台灣交貨時間多久?

※避免吃悶虧無故遭抬價!台中搬家公司免費估價,有契約讓您安心有保障!