大數據系列——Spark學習筆記Spark Streaming

1. Spark Streaming

• Spark Streaming是一個基於Spark Core之上的實時計算框架，能夠從不少數據源消費數據並對數據進行處理
• Spark Streaing中有一個最基本的抽象叫DStream（代理）,本質上就是一系列連續的RDD，DStream其實就是對RDD的封裝
• DStream能夠認爲是一個RDD的工廠，該DStream裏面生產都是相同業務邏輯的RDD，只不過是RDD裏面要讀取數據的不相同
• 在一個批次的處理時間間隔裏， DStream只產生一個RDD
• DStream就至關於一個"模板", 咱們能夠根據這個"模板"來處理一段時間間隔以內產生的這個rdd,以此爲依據來構建rdd的DAG

2. 當下比較流行的實時計算引擎

​ 吞吐量 編程語言 處理速度 生態

Storm 較低 clojure 很是快(亞秒) 阿里(JStorm)

Flink 較高 scala 較快(亞秒) 國內使用較少

Spark Streaming 很是高 scala 快(毫秒) 完善的生態圈

3. Spark Streaming處理網絡數據

//建立StreamingContext  至少要有兩個線程  一個線程用於接收數據  一個線程用於處理數據
val conf = new SparkConf().setAppName("Ops1").setMaster("local[2]")
val ssc = new StreamingContext(conf, Milliseconds(3000))
val receiverDS: ReceiverInputDStream[String] = ssc.socketTextStream("uplooking01", 44444)
val pairRetDS: DStream[(String, Int)] = receiverDS.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)
pairRetDS.print()
//開啓流計算
ssc.start()
//優雅的關閉
ssc.awaitTermination()

4. Spark Streaming接收數據的兩種方式(Kafka)

• Receiver

  • 偏移量是由zookeeper來維護的
  • 使用的是Kafka高級的API(消費者的API)
  • 編程簡單
  • 效率低(爲了保證數據的安全性,會開啓WAL)
  • kafka0.10的版本中已經完全棄用Receiver了
  • 生產環境通常不會使用這種方式

• Direct

  • 偏移量是有咱們來手動維護
  • 效率高(咱們直接把spark streaming接入到kafka的分區中了)
  • 編程比較複雜
  • 生產環境通常使用這種方式

5. Spark Streaming整合Kafka

• 基於Receiver的方式整合kafka(生產環境不建議使用,在0.10中已經移除了)

  //建立StreamingContext  至少要有兩個線程  一個線程用於接收數據  一個線程用於處理數據
  val conf = new SparkConf().setAppName("Ops1").setMaster("local[2]")
  val ssc = new StreamingContext(conf, Milliseconds(3000))
  val zkQuorum = "uplooking03:2181,uplooking04:2181,uplooking05:2181"
  val groupId = "myid"
  val topics = Map("hadoop" -> 3)
  val receiverDS: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)
  receiverDS.flatMap(_._2.split(" ")).map((_,1)).reduceByKey(_+_).print()
  ssc.start()
  ssc.awaitTermination()

• 基於Direct的方式(生產環境使用)

  //建立StreamingContext  至少要有兩個線程  一個線程用於接收數據  一個線程用於處理數據
  val conf = new SparkConf().setAppName("Ops1").setMaster("local[2]")
  val ssc = new StreamingContext(conf, Milliseconds(3000))
  val kafkaParams = Map("metadata.broker.list" -> "uplooking03:9092,uplooking04:9092,uplooking05:9092")
  val topics = Set("hadoop")
  val inputDS: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
  inputDS.flatMap(_._2.split(" ")).map((_, 1)).reduceByKey(_ + _).print()
  ssc.start()
  ssc.awaitTermination()

6. 實時流計算的架構

1. 生成日誌(模擬用戶訪問web應用的日誌)

public class GenerateAccessLog {
    public static void main(String[] args) throws IOException, InterruptedException {
        //準備數據
        int[] ips = {123, 18, 123, 112, 181, 16, 172, 183, 190, 191, 196, 120};
        String[] requesTypes = {"GET", "POST"};
        String[] cursors = {"/vip/112", "/vip/113", "/vip/114", "/vip/115", "/vip/116", "/vip/117", "/vip/118", "/vip/119", "/vip/120", "/vip/121", "/free/210", "/free/211", "/free/212", "/free/213", "/free/214", "/company/312", "/company/313", "/company/314", "/company/315"};

        String[] courseNames = {"大數據", "python", "java", "c++", "c", "scala", "android", "spark", "hadoop", "redis"};
        String[] references = {"www.baidu.com/", "www.sougou.com/", "www.sou.com/", "www.google.com"};
        FileWriter fw = new FileWriter(args[0]);
        PrintWriter printWriter = new PrintWriter(fw);
        while (true) {
            //            Thread.sleep(1000);
            //產生字段
            String date = new Date().toLocaleString();
            String method = requesTypes[getRandomNum(0, requesTypes.length)];
            String url = "/cursor" + cursors[getRandomNum(0, cursors.length)];
            String HTTPVERSION = "HTTP/1.1";
            String ip = ips[getRandomNum(0, ips.length)] + "." + ips[getRandomNum(0, ips.length)] + "." + ips[getRandomNum(0, ips.length)] + "." + ips[getRandomNum(0, ips.length)];
            String reference = references[getRandomNum(0, references.length)];
            String rowLog = date + " " + method + " " + url + " " + HTTPVERSION + " " + ip + " " + reference;
            printWriter.println(rowLog);
            printWriter.flush();
        }
    }


    //[start,end)
    public static int getRandomNum(int start, int end) {
        int i = new Random().nextInt(end - start) + start;
        return i;
    }
}

2. flume使用avro採集web應用服務器的日誌數據

• 採集命令執行的結果到avro中

# The configuration file needs to define the sources, 
# the channels and the sinks.
# Sources, channels and sinks are defined per agent, 
# in this case called 'agent'
f1.sources = r1
f1.channels = c1
f1.sinks = k1

#define sources
f1.sources.r1.type = exec
f1.sources.r1.command =tail -F /logs/access.log

#define channels
f1.channels.c1.type = memory
f1.channels.c1.capacity = 1000
f1.channels.c1.transactionCapacity = 100

#define sink  採集日誌到uplooking03
f1.sinks.k1.type = avro
f1.sinks.k1.hostname = uplooking03
f1.sinks.k1.port = 44444

#bind sources and sink to channel 
f1.sources.r1.channels = c1
f1.sinks.k1.channel = c1

• 從avro採集到控制檯

# The configuration file needs to define the sources, 
# the channels and the sinks.
# Sources, channels and sinks are defined per agent, 
# in this case called 'agent'
f2.sources = r2
f2.channels = c2
f2.sinks = k2

#define sources
f2.sources.r2.type = avro
f2.sources.r2.bind = uplooking03
f2.sources.r2.port = 44444

#define channels
f2.channels.c2.type = memory
f2.channels.c2.capacity = 1000
f2.channels.c2.transactionCapacity = 100

#define sink
f2.sinks.k2.type = logger

#bind sources and sink to channel 
f2.sources.r2.channels = c2
f2.sinks.k2.channel = c2

• 從avro採集到kafka中

# The configuration file needs to define the sources, 
# the channels and the sinks.
# Sources, channels and sinks are defined per agent, 
# in this case called 'agent'
f2.sources = r2
f2.channels = c2
f2.sinks = k2

#define sources
f2.sources.r2.type = avro
f2.sources.r2.bind = uplooking03
f2.sources.r2.port = 44444

#define channels
f2.channels.c2.type = memory
f2.channels.c2.capacity = 1000
f2.channels.c2.transactionCapacity = 100

#define sink
f2.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink
f2.sinks.k2.topic = hadoop
f2.sinks.k2.brokerList = uplooking03:9092,uplooking04:9092,uplooking05:9092
f2.sinks.k2.requiredAcks = 1
f2.sinks.k2.batchSize = 2

#bind sources and sink to channel 
f2.sources.r2.channels = c2
f2.sinks.k2.channel = c2