專題

謝謝吉豫的報告，接下來由我接手演示，如何使用這個系統，來將收集到的RAW Data 彙整成對企業有價值的資料。

這個過程總共有分三個步驟，

第一步，因為時間的關係，我先執行了這段命令，執行的結果可以看到這些都是google sheet上面的RAW Data，由這裡的時間可以證明剛剛吉豫所填寫google sheet的表單， 會透過connector連結到kafka系統裡面，並且即時彙整到client_sent_QA這個Topic。

第二步，彙整成client_sent_QA後，透過執行我們撰寫的python程式，可以看到這筆資料的時間、公司跟剛剛從Topic回傳的是一樣的，後面新增的這3欄，第一欄就是從open api查回來的正確的公司名稱，第二欄顯示是否為新舊客戶，第三欄就是當時公司發派的接洽業務是誰。

new是新客戶，因為是新客戶，以na來表示待指派業務的狀態。

以上比對結果，會再彙整到client_QA_extend這個Topic。

最後，我們在HDFS檔案系統裡，檢視client_QA_extend這個Topic的檔案，是否有存在HDFS檔案系統裡面?

有

看到檔案名稱後，再執行cat來檢視這個檔案，是否與client_QA_extend這個Topic裡面的資料一致。

OK一樣

也就是說client_QA_extend裡的資料會透過HDFS 2 Sink Connector， 自動連結到HDFS檔案系統裡做延伸應用，成為有商業價值的資料。

我的演示到此結束，接下來請昶瑋替我們報告效益與延伸應用。

client_sent_QA // client_QA_extend

吉豫剛剛在google sheet新增的資料，我們可以從

dtw1主機 192.168.207.163

要看google sheet的資料，指令如下: kafkacat -b dtw1 -t client_sent_QA | jq .

dtm2主機 192.168.207.166

看python程式，指令如下: python3 client.py

我們的python程式會去抓mariadb的資料以及政府的open data裡面的數據，來核對剛剛在client_sent_QA這個Topic回傳的資料一樣，可以看到後面多了3個欄位，第一欄是從open api查回來的資料，第二欄的new是以前不存在我們的客戶系統中，最後的na是因為他是新客戶，所以還沒指派過業務，所以用na表示帶指派業務的狀態。

dtm1主機 192.168.207.161

• 看Hadoop資料的路徑，指令如下: hdfs dfs -ls /topics/client_QA_extend/partition=0

• 看到資料的檔案名稱後，指令如下: hdfs dfs -cat /topics/client_QA_extend/partition=0/檔名

• 更改connector每"1"筆紀錄1次，先到他的設定檔，指令如下: nano hdfs2.properties 設定第5行的flush.size=1，改完後，跳出去輸入: jps 再輸入 kill -9 (ConnectStandalone的pid) 要先把connector關掉後再重啟，重啟指令: connect-standalone.sh /opt/kafka_2.13-3.0.0/config/connect-standalone.properties hdfs2.properties &

dtm2主機 192.168.207.166

啟connector的server時候，其中一個設定檔，指令如下: nano /opt/kafka_2.13-3.0.0/config/connect-standalone.properties