本文是我翻譯自redis官方網站文章，英文作者是Itamar Harber。

Spark-Redis是用Spark在redis上面進行讀寫資料操作的包。其支援redis的所有資料結構：String（字串）, Hash（雜湊）, List（列表）, Set and Sorted Set（集合和有序集合）。此模組既可以用於Redis的standalone模式，也可用於叢集情況。此外，Spark-Redis還提供了對Spark-Streaming的支援。

簡介

本文概括了開始使用Apache Spark和Redis所需的基本步驟。關於Spark和Redis的安裝暫不介紹。我們將使用“WORD COUNT”為例，來介紹Spark，Redis和spark-redis的聯合使用。
 

Redis實驗室最近於2015年9月釋出了spark-redis package。顯而易見，根據它的名字，這是一個為Apache Spark提供Redis連線的連線件，它允許人們對Redis的資料結構在Spark中以RDD（彈性分散式資料集，Spark的專門術語。）的結構形式進行操作。

自從Spark開源以來，由於其針對大規模資料處理的高效且通用的引擎（輕鬆的超過了之前只能在單一平臺上面操作的大資料平臺），很快吸引了開發者們的注意。Spark採用了迴圈資料流和記憶體計算，使得其比Hadoop的MapReduce速度快了很多倍。由於Spark的易用性和SQL，Streaming以及Mlib等庫的擴充套件，吸引了開發人員的眼球。

Redis將共享記憶體的架構（shared in-memory infrastructure）引入到Spark中，這使得Spark處理資料的速度又快了幾個數量級。此外，Redis的資料結構簡化了資料的獲取和處理，使程式碼複雜度下降，並節省了網路通訊和頻寬的消耗。

因此，兩者的結合，可以實現大規模資料量的實時處理任務。提速幅度有多大?如果Redis和Spark結合使用，結果證明，處理資料(以時間序列資料為例)的速度比Spark單單使用程序記憶體或堆外快取來儲存資料要快45倍――不是快45%，而是快整整45倍!

配置

• Apache Spark
• Scala
• Jedis
• Redis

最低標準：
Apache Spark v1.4.0
Scala v2.10.4
Jedis v2.7
Redis v2.8.12 or v3.0.3

Example（Word Count計數器）

下面，我們使用word count例子來開始介紹Spark-Redis的使用。

Step 1：讀取資料

這裡，我們將對Redis原始碼檔案進行Word Count統計，希望得出一些有趣的結論。當上面配置好以後，我們執行：

注意：$ bin/spark-shell --jars <path-to>/spark-redis-<version>.jar,<path-to>/jedis-<version>.jar
需要在–jars 後面加上這兩個jar包。

在這裡，輸入：

wtext.count = 100

表示有100個Redis的原始檔。當然，你也可以用ls -l redis/src/*.[ch] | wc -l 來統計檔案個數。但是通過WholeTextFileRDD的方式，可以看出job的劃分stage和完成的情況。

Step 2：改變檔案內容

接下來，是將檔案變成一個個單詞。相比於普遍使用的TextFileRDD,WholeTextFileRDD返回一個key-value型的結構資料。key對應的是每個檔案的path；value對應的是檔案的內容。
下面是WholeTextFileRDD的程式碼。

/** 
   * Read a directory of text files from HDFS, a local file system (available on all nodes), or any 
   * Hadoop-supported file system URI. Each file is read as a single record and returned in a 
   * key-value pair, where the key is the path of each file, the value is the content of each file. 
   * 
   * <p> For example, if you have the following files: 
   * {{{ 
   *   hdfs://a-hdfs-path/part-00000 
   *   hdfs://a-hdfs-path/part-00001 
   *   ... 
   *   hdfs://a-hdfs-path/part-nnnnn 
   * }}} 
   * 
   * Do `val rdd = sparkContext.wholeTextFile("hdfs://a-hdfs-path")`, 
   * 
   * <p> then `rdd` contains 
   * {{{ 
   *   (a-hdfs-path/part-00000, its content) 
   *   (a-hdfs-path/part-00001, its content) 
   *   ... 
   *   (a-hdfs-path/part-nnnnn, its content) 
   * }}} 
   * 
   * @note Small files are preferred, large file is also allowable, but may cause bad performance. 
   * @note On some filesystems, `.../path/*` can be a more efficient way to read all files 
   *       in a directory rather than `.../path/` or `.../path` 
   * 
   * @param path Directory to the input data files, the path can be comma separated paths as the 
   *             list of inputs. 
   * @param minPartitions A suggestion value of the minimal splitting number for input data. 
   */  
  def wholeTextFiles(  
      path: String,  
      minPartitions: Int = defaultMinPartitions): RDD[(String, String)] = withScope {  
    assertNotStopped()  
    val job = NewHadoopJob.getInstance(hadoopConfiguration)  
    // Use setInputPaths so that wholeTextFiles aligns with hadoopFile/textFile in taking  
    // comma separated files as input. (see SPARK-7155)  
    NewFileInputFormat.setInputPaths(job, path)  
    val updateConf = job.getConfiguration  
    new WholeTextFileRDD(  
      this,  
      classOf[WholeTextFileInputFormat],  
      classOf[Text],  
      classOf[Text],  
      updateConf,  
      minPartitions).map(record => (record._1.toString, record._2.toString)).setName(path)  
  }  

值得注意的是：① wholeTextFiles對於大量的小檔案效率較高，大檔案效果不太好。
② 一些檔案系統的路徑名採用萬用字元的形式效果比一個一個檔名新增上去更高效。

將檔案變成（檔名：單詞）的形式。（變數的名稱：wtext代表WholeTextFiles, fwd代表FileWords。）：

當fwds這個RDD將檔名和單詞清楚的分開之後，我們就準備好進行詞頻統計了。在此之前，先對所有檔案的單詞進行彙總，進行一個整體的詞頻統計。

其結果為：

Step 3：將RDD寫到Redis中

從這步開始，我們就開始使用到非常6的Redis了！我們將用Redis儲存這個結果，以供後續計算使用。Redis的有序集合對詞頻統計(單詞：出現次數)場景特別適合，既可以根據單詞找到單詞的出現次數，也可以通過單詞的出現次數找到符合要求的單詞。

實現這個只需要一行程式碼！！！

一旦資料存放到Redis中，我們可以使用命令列操作：

Step 4：從Redis中讀取RDD

相較於從Redis寫資料，更常見的場景是從Redis中讀資料。執行下面的程式碼，使得每一個檔案的詞頻統計彙總起來變成一個總的輸出。

用spark-shell進行操作，獲取全部單詞的個數（包括重複的內容）。

閉注(Closing Notes)

當資料量小的時候，我們統計單詞個數可以用wc -w來實現。隨著資料量的增大，需要找到新的方法來抽象解答方式並增加方法的靈活性和可擴充套件性。Spark是一個讓人興奮的大資料處理工具。更別說其和Hadoop生態系統耦合，並有諸如SQL，streaming，Mlib等擴充套件包。
Redis的出現，可以說是為Spark“ 解了渴“。spark-redis
通過簡單的幾行程式碼，將RDD和Redis的核心資料結構迅速互換。spark-redis包已經提供了直接的方式將RDD和redis的結構進行互換，並提供了友好的方式來獲取key的名字。此外，連線件還通過將RDD分割槽轉換成Redis的hash slot，有效的減少了引擎內部shuffling操作。

最後，這個開源的連線件還在不斷的發展中，將來有可能會被spark設定成預設元件。

參考(References)