一直在研究大資料處理方向的各種技術和工具，但沒有真正用於實踐中，恰好目前風控措施轉向，需要實施“線上+離線”的雙線防控措施，遂在調研查閱相關文件後，決定從零開始構造（資料探勘轉工程開發，思維轉變是關鍵），這裡面涉及的幾個階段慢慢說來：

1. 專案開發環境選擇（scala2.10.4 IDE +maven3.3）
   最開始的選擇是直接在eclipse環境上安裝scala的IDE外掛，但實踐證明此種方式很不穩定，受限於網路的限制基本沒能成功，索性直接利用scala提供的IDE開發環境，這視為邁出第一步，雖然IDE提供了構建scala project的模板，但是後期證明沒有合適的版本管理工具，很難在包依賴及部署上做到得心應手，雖然網上普遍推薦SBT在該scala專案管理上的作用，但是有過java開發經驗的還是首推maven工具，為編譯打包提供極大的方便，同時實現在windows環境下編譯執行spark程式碼，不過裡面涉及很多坑後面再術。

2. 實際計算平臺（spark 1.5.2）
   網上關於storm與spark ，誰在流式計算方式更具有優勢的討論甚多，這裡不做比較，本人結合已有的平臺環境加開發工具選擇了 spark streaming作為實時計算的計算引擎，另外一個原因在於spark在於機器學習支撐上的強有力地位，方便後日擴充套件，此外spark streaming也提供了針對各種資料來源的高階API，方便從不同資料來源中獲取DStearm，同時支援寫入各種儲存介質中。

3. 資料來源（kafka）
   kafka作為一個分散式釋出-訂閱訊息系統，可以利用已有的API實時的從topic中pull資料，滿足實時計算的要求（前期嘗試從hdfs中讀取資料做計算並非有效），同時streaming-kafka的API友好的提供了資料來源無縫的讀取方式，一切都為達到實時計算的目的

4. 資料去向（HDFS）

利用上述提到的各種平臺工具等， 舉個簡單案例，程式碼如下：

package com.kafka.test

import java.util.HashMap
import org.apache.kafka.clients.producer.{ProducerConfig, KafkaProducer, ProducerRecord}
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf

object KafkaWordCounts {
    def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    return Some(newValues.sum
 
+runningCount.getOrElse(0))
  }

  def main(args: Array[String]) {
    //System.setProperty("hadoop.home.dir", "E:\\software\\hadoop-2.5.2");
     val zkQuorum = "101.271.251.161:2181"
     val group = "zjz-test-waf8"
     val topics = "monitor_firewall_accesslog_pre"
     val numThreads="2"
     if (args.length >= 4){
        val zkQuorum = args(0)
        val group = args(1)
        val topics = args(2)
        val numThreads=args(3)
     }
    val sparkConf = new SparkConf().setAppName("KafkaWordCount")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    ssc.checkpoint("checkpoint")

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
    val words = lines.flatMap(_.split("\\s+"))
    val wordCounts = words.map(x => (x, 1)).updateStateByKey[Int](updateFunction(_,_))
    wordCounts.print()
    wordCounts.saveAsTextFiles("sstest/kafka_waf_streaming","txt")
    ssc.start()
    ssc.awaitTermination()
  }
}