面試（串講三）
十二、Hbase
12.1 基本架構
12.2 讀寫流程
寫流程：
1）Client先訪問Zookeeper，獲取hbase:meta表位於哪個RegionServer。
2）訪問對應的RegionServer，獲取hbase:meta表，根據讀請求的namespace:table/rowkey，查詢出目標資料位於哪個RegionServer中的哪個Region中。並將該table的region資訊以及meta表的位置資訊快取在客戶端的meta cache，方便下次訪問。
3）與目標RegionServer進行通訊；
4）將資料順序寫入（追加）到WAL（HLog日誌檔案）；
5）將資料寫入對應的MemStore，資料會在MemStore進行排序；
6）向客戶端傳送ack；
7）等達到MemStore的刷寫時機後，將資料刷寫到HFile。
讀流程：
1）Client先訪問Zookeeper，獲取hbase:meta表位於哪個RegionServer。
2）訪問對應的RegionServer，獲取hbase:meta表，根據讀請求的namespace:table/rowkey，查詢出目標資料位於哪個RegionServer中的哪個Region中。並將該table的region資訊以及meta表的位置資訊快取在客戶端的meta cache，方便下次訪問。
3）與目標RegionServer進行通訊；
4）分別在BlockCache（讀快取），MemStore中查詢目標資料，如果BlockCache中未查到相應資料則掃描對應的HFile檔案，HFile中掃描到的資料塊（預設64K）寫入BlockCache，並將查到的所有資料進行合併。此處所有資料是指同一條資料的不同版本（timestamp）或者不同的型別（Put/Delete）。
5）將合併後的最終結果返回給客戶端。
刷寫條件：
memstore級別： flushsize ，預設 128m
region級別： 4 * flushSize ，觸發刷寫，預設512m
region server 級別： 0.4 * 堆記憶體 * 0.95
hlog ： 32個檔案數
手動刷寫： 執行命令
12.3 合併
major：大合併，所有的Hfile合併成一個大的Hfile，會將過期、標記為刪除的資料清理掉
minor：小合併，把相鄰的幾個Hfile合併成一個較大的Hfile，不會將過期、標記為刪除的資料清理掉
企業怎麼用？
大合併影響提供服務，預設7天一次
調整時間
關閉自動大合併，自己選擇時間進行大合併
保持預設（我們公司）
12.4 自動切分
0.94：按照10G切分
0.94-2.0：min(2 * R^3 * flushSize，10G) 注：R表示某個RegionServer下的Region數
第一次按256M切，閾值越來越大，直到超過10G，後面一直10G切
2.0之後：第一次按256M切，後面全部按10G切
12.5 rowkey設定原則
1）唯一原則：不重複
2）長度原則：最長64kb，一般設為10~100b，最好是16B
3）雜湊原則：資料分佈均勻
取Hash、加鹽、反轉、隨機數
原始資料：1234567_ts
取Hash：hash(1234567_ts)
最終的rowkey：hash(1234567_ts)_1234567_ts
rowkey的比較：按照字典順序，按位比較
12.6 分割槽鍵
建表時，進行預分割槽，SPLITS =》 {'000|','001|','002|'}
-∞~ 000|
000| ~ 001| ==》 001234312521
001| ~ 002| ==》 002345422
002| ~ +∞
12.7 熱點問題
請求集中在個別的RegionServer上
解決思路：
將資料打散
良好的rowkey設計 + 合理的分割槽鍵
比如：1234567_ts
取隨機數：1234567_ts --> a
再取模，按分割槽數取模：a % 4 =b (0、1、2、3)
最後拼接到原始的rowkey前：00b_1234567_ts
12.8 優化
1）記憶體：16~48G，我們公司設定為40G
2）9個引數：調大writebuffer、追加寫、調大超時時長...
3）rowkey + 分割槽 ！！！！
12.9 二級索引
rowkey zs 18 中性
實現：藉助Phoenix
全域性索引：索引表、資料表，在不同的region
create index 索引表名 on 資料表名（索引列）;
索引表的rowkey = 索引列_原來的rowkey
本地索引：索引資料、資料表在同一個region
create local index 索引表名 on 資料表名（索引列）;
索引資料的rowkey = 分割槽鍵_索引列_原始rowkey
資料直接寫入HBase，不通過phoenix，新的資料會建立二級索引嗎？
只能通過Phoenix寫的資料，才會新增二級索引
如何解決？
往phoenix再寫一次
十三、Spark
13.1 入門
1）常用埠號
4040：執行中的job的ui頁面
18080：spark的歷史伺服器
7077：老大的rpc埠
2）部署模式
standalone：自己管資源
yarn：yarn管資源
client：driver啟動在本地
cluster：driver位置由yarn決定
k8s：
mesos：國外
3）配置檔案
spark-default.conf
spark-env.conf
13.2 core
1）RDD是什麼？五大屬性
分散式彈性資料集
計算分割槽
計算邏輯
血緣依賴
分割槽器
移動資料不如移動計算
2）RDD可變嗎？RDD儲存資料嗎？
不可變，只存元資料，不儲存真正的資料
3）共享變數
累加器
廣播變數：存在Executor的BlockManger上
4）cache、checkpoint
cache：基於記憶體，不會切斷血緣關係
checkpoint：基於HDFS，會切斷血緣關係
開發中怎麼用？cache + checkpoint
5）spark的checkpoint與flink的checkpoint有什麼區別？
spark的checkpoint，只儲存了driver的元資料，只是一個簡單的狀態
flink的checkpoint，使用了Chandy-Lamport演算法，非同步分界線快照演算法
6）運算元
單value：
map、flatmap、filter、groupby、mapPartition
雙value：rdd1.xxx(rdd2)
union、zip、join、intersection、subtract
k-v：
groupbykey
sortbykey：對key進行排序
如果想對value排序，使用map運算元調換一下順序，然後再sortkeyby，最後再使用map換回來
| 初始值 | 分割槽內和分割槽間邏輯是否一致 | |
| reducebykey | 無 | 一致 |
| foldbykey | 有值 | 一致 |
| aggregatebykey | 有值 | 可以不一致 |
| combinebykey | 有函式 | 可以不一致 |
xxxbykey：會產生shuffle
手動重分割槽：
repartition：一般用來增加分割槽，底層實現是使用coalesce，一定會產生shuffle
coalesce：一般用來減少分割槽，不一定會產生shuffle
行動運算元：foreach、collect、count、take、last、first、reduce、save
13.3 spark幾種劃分
job：遇到一個行動運算元，就會生成一個job
stage：遇到shuffle（寬依賴），就會劃分stage
taskset：一個stage中，task的集合
task：一個stage中，最後一個rdd的分割槽數
13.4 寬窄依賴
寬依賴產生shuffle，窄依賴不產生shuffle
13.5 spark提交流程，通訊流程、任務排程、記憶體模型（統一記憶體模型）
13.6 spark優化
13.7 提交引數
spark-submit #指定模式 --master yarn --deploy-mode cluster #指定driver的資源 --driver-cores --driver-memory #指定Executor的資源 --num-executors --executor-cores --executor-memory --class xxx.xxx.xxx. jar包 main方法的傳參
13.8 spark sql
1）Hive On SPark vs Spark On Hive
Hive On Spark ： 只有執行引擎換成了spark，其他的 解析、編譯、優化都是Hive自己來
Spark On Hive ： 只是用了Hive的元資料，其他的 解析、編譯、優化都是Spark自己來
2）有幾種抽象？
rdd、df、ds
3)、spark sql中， 大小join怎麼做最好？
廣播小表
4）sparksql 有幾種join？
broadcast hash join
shuffled hash join
sort merge join
Hive有幾種join？
mapjoin
common join （shuffle join、reduce join）
SMB join
5）Spark SQL 怎麼指定操作hive表？
建立 sparksession時， 呼叫 enableHiveSupport()，SparkSession.builder().xxx.xxxx.enableHiveSupport().getOrCreate()
6）小檔案：
呼叫 coalesce，調整輸出前的分割槽數
再啟一個job，定期去合併小檔案
13.9 spark streaming
1）計算模型？
微批次，每一個批次就是一個RDD
2）視窗 ：視窗長度、滑動步長
滾動： 視窗長度 = 滑動步長
滑動： 視窗長度 > 滑動步長
視窗長度 = 批次間隔的 整數倍
3）常用引數
優雅關閉
背壓
限流： 消費kafka， 單分割槽的速率上限
十四、Flink
14.1 入門
1）程式設計模型：分層API、程式設計四部曲（ENV、Source、Transform、Sink）
2）部署模式
standalone：自己管資源
yarn：yarn管資源
pre-job
yarn-session
application
k8s
mesos
企業怎麼選？建議pre-job、application為主
3）常用埠號
6123 JobManager RPC埠
8081 Standalone模型下的webUI，若是yarn模式，通過8088跳轉
4）配置檔案
5）世界觀：一切皆是流
14.2 基礎
1）ENV：底層自動幫我們判斷環境
2）Source：env.addSource
kafka
source之後就生成水印
設定空閒等待
3）transform：
map、flatmap、filter、connect、union、shuffled、rebalance、rescale
keyby：同一個分組的資料，會進入同一個分割槽
同一個分割槽內，可以有多個分組
兩次Hash：第一次key本身的hashcode方法-->hash1
第二次hash-->hash2，對最大的並行度取模，得到鍵組ID（0~127）
計算分割槽：鍵組ID * 下游運算元並行度 / 最大並行度
connect：一次只能連線兩條流，型別可以不一致
union：一次可以合併多條流，型別必須一致
interval join：
（1）判斷是否遲到，遲到就不處理
（2）每條流，各自存了Map型別的狀態，來一條資料就存進去
key=時間戳，value=資料的集合
（3）不管哪條流的資料來了，它都會去對方的Map狀態裡遍歷，檢視是否能匹配上
（4）超過區間時，會清理Map對應的資料
4）sink：流.addSink()
14.3 高階
1）時間語義
處理時間
事件時間
注入時間
2）談談你對WaterMark的理解？
衡量事件時間的進展
處理亂序、遲到資料
單調不減
是一個特殊的時間戳，往下游傳遞
觸發視窗、定時器的計算
Flink認為，時間戳小於watermark的資料，應該都處理過了，如果後面還有，則該資料為遲到資料
3）Watermark的傳遞
一對多：廣播
多對一：以最小的為準
多對多：拆開來看
4）watermark是週期生成的嗎？週期多久？
預設是週期生成，預設200ms
watermark = maxTs - 亂序程度 - 1ms
亂序程度：抽樣估算、經驗值（秒級或分鐘級）
5）視窗分類
時間：滾動、滑動、會話
條數：滾動、滑動
6）視窗怎麼劃分？
視窗開始時間 = 事件時間戳 對 視窗長度取整
視窗結束時間 = 開始時間 + WindowSize
watermark >= 視窗最大時間戳，觸發視窗的輸出
7）視窗為什麼左閉右開？
屬於視窗的最大時間戳 = end - 1ms
8）什麼時候建立、關窗（銷燬）？
屬於本視窗的第一條資料來的時間，現new的
watermark >= 視窗最大時間戳 + 允許遲到的時間
9）Flink怎麼保證一致性？（丟數、重複）
狀態：運算元狀態、鍵控狀態（值、map、list、聚合類）、廣播狀態
狀態後端：主要負責兩件事，一是本地狀態的管理，二是checkpoint的資料管理
| 本地狀態 | checkpoint資料 | |
| 記憶體 | TM的堆記憶體 | JM的堆記憶體 |
| FS | TM的堆記憶體 | HDFS |
| RocksDB | RocksDB | HDFS |
企業怎麼選？ 一般用FS， 大廠用 RocksDB
端到端一致性的概念：
資料來源：可重置，kafka滿足要求
計算引擎：flink內部的checkpoint機制，barrier對齊
輸出系統：冪等性、事務，輸出是kafka，符合條件
展開checkpoint演算法、barrier對齊
寫到檔案系統：
開啟事務-->do--->do--->提交事務，如果失敗就回滾
寫入臨時檔案，成功，更改為正式檔名
失敗，將臨時檔案刪除
10）Flink怎麼處理亂序、遲到？
（1）watermark
（2）允許遲到（視窗中設定）
（3）側輸出流
14.4 CEP&SQL
CEP三部曲：定義規則、應用規則、獲取結果
SQL：動態表、連續查詢 -->SQL解析器，Calcite
14.5 優化
1）提交引數：資源
2）如果是rocksdb：一些引數的調整
3）怎麼處理反壓？
（1）先斷鏈
（2）webui看：第一個為ok的運算元
（3）看對應運算元的程式碼邏輯，修改
4）怎麼處理資料傾斜？
（1）keyby前：利用重分割槽的運算元，rebalance、rescale、shuffle
（2）keyby後：
（3）開窗
5）SQL引數
十五、實時數倉
15.1 ods層
使用者行為日誌、MySQL業務資料（maxwell）
2個Topic
15.2 dwd層 + dim層
kafka消費有序性（使用event time）
日誌資料：頁面日誌放主流，其他4個放側輸出流（5個topic，頁面，啟動，曝光，事件，錯誤）
業務資料：
事實表：存kafka
維度表：存HBase、支援隨時查詢、長遠考慮，沒有做維度退化
動態分流：
首先在MySQL中建立一張配置表，指定了哪些表要去往哪裡
然後使用FlinkCDC讀取配置表，廣播出去，做成廣播流的形式
然後將廣播流與主流中的資料進行connect，每一條資料都會匹配到一個配置資訊，為接下來分流做準備
根據每條資料的配置分流出事實表和維度表，事實表根據配置資訊中的Table名稱作為Topic，然後sink到該Topic中
維度表先查詢該表是否已經建立，如果沒有建立，則先建立表（使用Phoenix），然後再往對應的表中寫入資料
15.3 dwm層：預先處理的寬表
過渡層：避免拉寬表的時候，重複計算
事實表與事實表關聯：雙流Join（interval join）
事實表與維度表關聯：通過Phoenix查詢HBase
使用了非同步IO的方式
做了旁路快取（二級快取）：加速查詢
如何保證快取一致性？
如果是更新HBase，主動將Redis的資料進行一個更新處理
redis設定了過期時間，24小時（避免冷資料，長期佔用快取）
15.4 dws層：拉寬表（join和查維度）
寬表存ClickHouse
為什麼存ClickHouse，為什麼不存HBase？
HBase適合存明細資料，對於聚合分析統計，不太擅長，效率不高
ClickHouse聚合分析統計，快，不擅長join，適合放寬表這種已經關聯過的資料表，欄位多、資料量大的
開了一個小視窗：攢批，攢批一次性寫入ClickHouse（因為ClickHouse對併發支援不好，最高每秒100次請求）
15.5 怎麼保證一致性？
寫Kafka：flink的checkpoint、兩階段提交的sink
寫HBase：冪等性
寫ClickHouse：ReplacingMergeTree（合併樹），但只能保證最終一致性，合併分片時才會去重
解決方案：
1）手動優化
2）查ClickHouse時，加個group by
3）使用final（去重並取最新的資料），早期版本只支援單執行緒，20.5開始才開始支援多執行緒
使用方式：在表名後面加final即可
怎麼用？推薦使用final + group by的方式
15.6 ads層
直接實時查詢 dws層的資料，展示給前端（sugar）
15.7 一共幾個Topic？
ods：2個
dwd：日誌5個、業務15個
dwm：4個
dws：4個
2 + 5 + 15 + 4 + 4 = 30，所以我們公司有30多個Topic
15.8 一共幾張表？
業務表30多張
15.9 資料量、速率、條數？
日活60萬，平均每人產生100條資料，每條資料大小在0.5k~2k之間，平均取1k，所以每天產生60多G左右的資料
平均每秒產生700多條資料，平均每秒產生0.7M左右的資料
高峰期每秒產生1萬多條資料，每秒產生10多M的資料
低谷期每秒產生40多條資料，每秒產生40k左右的資料
15.10 用什麼監控？
Kafka：eagle
Flink：Prometheus + Grafana
ClickHouse：Prometheus + Grafana
15.11 ClickHouse支不支援更新？
是一個OLAP資料庫，不支援事務
能更新、刪除，但不是真正的更新和刪除，使用：alter table xxx 實現
更新、刪除的資料，只有在分片時才會真正的清理