概述:本文主要介紹spark streaming的幾個應用案例。
1、統計結果寫入資料庫
(1)建立資料庫
create table wordcount(
word varchar(50) default null,
wordcount int(10) default null
);
(1)建立連線
def createconnection() =
(3)**實現
def main(args: array[string]): unit =
// }
result.print()
result.foreachrdd )
connection.close()}}
ssc.start()
ssc.awaittermination()
}
2、黑名單過濾
(1)功能分析
#訪問日誌 ==>dstream
20190516,zhangsan
20190516,lisi
20190516,wangwu
20190516,zhaoliu
資料轉換==>(zhangsan:20190516,zhangsan),(lisi:20190516,lisi), (wangwu:20190516,wangwu),(zhaoliu:20190516,zhaoliu)
#黑名單 ==>rdd
lisi
zhaoliu
資料轉換==>(lisi:true),(zhaoliu,true)
==>輸出
20190516,zhangsan
20190516,wangwu
leftjoin
(zhangsan:<20190516,zhangsan>,),(lisi:<20190516,lisi>,), (wangwu:<20190516,wangwu>,),(zhaoliu:<20190516,zhaoliu>,)
(2)**實現
def main(args: array[string]): unit = )
clicklog.print()
scc.start()
scc.awaittermination()
}}3、spark stream與spark sql操作實現字元統計
(1)匯入依賴
org.apache.spark
spark-sql_2.11
2.4.2
(2)**實現
object sqlnetworkwordcount
ssc.start()
ssc.awaittermination()
}case class record(word: string)
/** lazily instantiated singleton instance of sparksession */
object sparksessionsingleton
instance}}
}
統計結果
Spark Streaming入門詳解
背景 使用spark主要是使用spark streaming,spark streaming的魔力之所在於 1.流式處理,如今是乙個流處理時代,一切與流不相關的都是無效的資料。3.spark streaming本身是乙個程式,spark streaming在處理資料的時候會不斷感知資料。所以對構建複...
Spark Streaming 程式監控
官網中指出,spark中專門為sparkstreaming程式的監控設定了額外的途徑,當使用streamingcontext時,在web ui中會出現乙個 streaming 的選項卡,在此選項卡內,統計的內容展示如下 這其中包括接受的記錄數量,每乙個batch內處理的記錄數,處理時間,以及總共消耗...
spark streaming讀取kafka示例
spark streaming讀取kafka示例,其中 spark streaming優雅的關閉策略優化部分參考 如何管理spark streaming消費kafka的偏移量部分參考 spark向kafka中寫入資料部分參考 object demo 建立streamingcontext return...