大數(shù)據(jù)開發(fā)項目實戰(zhàn)-PPT課件第4章 廣電大數(shù)據(jù)用戶畫像-數(shù)據(jù)采集與預(yù)處理

廣電大數(shù)據(jù)用戶畫像——

數(shù)據(jù)采集與預(yù)處理1數(shù)據(jù)存儲與傳輸目錄業(yè)務(wù)數(shù)據(jù)2基礎(chǔ)數(shù)據(jù)預(yù)處理3在實際的生產(chǎn)環(huán)境中，用戶的業(yè)務(wù)系統(tǒng)記錄了用戶的基本信息、每個月用戶的賬單信息、用戶的訂單信息、用戶狀態(tài)變更的信息和用戶收視行為的信息。系統(tǒng)每天會定時將前一天業(yè)務(wù)系統(tǒng)產(chǎn)生的數(shù)據(jù)增量同步到Elasticsearch集群中，以服務(wù)其他業(yè)務(wù)系統(tǒng)，因此，在本項目中是需要從Elasticsearch集群讀取數(shù)據(jù)的。在用戶的Elasticsearch集群中，用戶基本信息表、用戶狀態(tài)信息變更表、賬單信息表、訂單信息表這4張表的數(shù)據(jù)在Elasticsearch集群中各自用一張表來存儲，而用戶收視行為信息表的數(shù)據(jù)較多，因此將用戶收視行為信息表的數(shù)據(jù)以周為單位進行存儲。生產(chǎn)數(shù)據(jù)來源Elasticsearch集群各節(jié)點的角色模擬數(shù)據(jù)產(chǎn)生1．Elasticsearch安裝配置節(jié)點IP地址角色node15既作為data節(jié)點也能夠被選舉為master節(jié)點node26既作為data節(jié)點也能夠被選舉為master節(jié)點node37既作為data節(jié)點也能夠被選舉為master節(jié)點Elasticsearch安裝配置的具體步驟在node1節(jié)點上下載并解壓elasticsearch-6.3.2.tar.gz，并修改elasticsearch-6.3.2/config/elasticsearch.yml文件。配置/etc/security/limits.conf文件，修改/etc/security/limits.d/90-nproc.conf文件。創(chuàng)建一個elasticsearch用戶，并更改elasticsearch-6.3.2目錄的所屬用戶和所屬組為elasticsearch。從root用戶切換為elasticsearch用戶，再啟動Elasticsearch。模擬數(shù)據(jù)產(chǎn)生Elasticsearch安裝配置的具體步驟瀏覽器的地址欄訪問“5:9200”。其他節(jié)點的安裝，可以復(fù)制當(dāng)前節(jié)點的整個elasticsearch-6.3.2目錄到其他節(jié)點中，只需要修改elasticsearch.yml的和network.host的值即可，其他步驟是相同的。模擬數(shù)據(jù)產(chǎn)生Elasticsearch的head插件是一種界面化的集群操作和管理工具，用來輔助管理Elasticsearch集群。Elasticsearch的head插件安裝在node1節(jié)點上，使用默認(rèn)的端口9100。瀏覽器訪問head插件的地址“5:9100”。模擬數(shù)據(jù)產(chǎn)生因為乙方人員拿到的數(shù)據(jù)是CSV格式的文件數(shù)據(jù)，所以需要先將CSV格式的文件數(shù)據(jù)導(dǎo)入Elasticsearch集群，可以使用Logstash將文件數(shù)據(jù)導(dǎo)入Elasticsearch集群。Logstash是開源的服務(wù)器端數(shù)據(jù)處理管道，能夠同時從多個數(shù)據(jù)來源采集數(shù)據(jù)、轉(zhuǎn)換數(shù)據(jù)，并將數(shù)據(jù)發(fā)送到指定的數(shù)據(jù)存儲庫中。Logstash版本選擇與Elasticsearch對應(yīng)的版本6.3.2。Logstash的安裝比較簡單，解壓即可使用，此處在server1（IP地址為3）上安裝了Logstash。模擬數(shù)據(jù)產(chǎn)生2．?dāng)?shù)據(jù)導(dǎo)入ElasticsearchLogstash運行時需要指定一個配置文件，配置文件中可以指定輸入（input）、過濾（filter）及輸出（output）這3個組件的插件類型。Logstash在輸入、過濾和輸出管道中都支持多種類型的插件，需要根據(jù)實際需求來選擇合適的插件。因為給定的數(shù)據(jù)源是CSV文件格式的，并且最終數(shù)據(jù)需要存儲到Elasticsearch集群中，所以選擇輸入插件的類型為file，過濾插件的類型為CSV，輸出插件的類型為Elasticsearch。模擬數(shù)據(jù)產(chǎn)生CSV數(shù)據(jù)導(dǎo)入Elasticsearch集群的流程模擬數(shù)據(jù)產(chǎn)生input的file插件的部分參數(shù)說明模擬數(shù)據(jù)產(chǎn)生filter的CSV插件的部分參數(shù)說明output的Elasticsearch插件的部分參數(shù)說明模擬數(shù)據(jù)產(chǎn)生在Logstash的安裝目錄中，創(chuàng)建custom_config目錄，在custom_config目錄下創(chuàng)建Logstash的配置文件mediamatch_usermsg.conf，實現(xiàn)導(dǎo)入mediamatch_usermsg.csv數(shù)據(jù)到Elasticsearch集群中。將mediamatch_usermsg.csv文件上傳到server1（IP地址為192.168.111.73）的/opt/data/usermsg目錄下。在Logstash的安裝目錄下創(chuàng)建sincedb_path目錄，并在sincedb_path目錄下創(chuàng)建userevent.sdp文件，用于保存讀取文件進度的記錄。啟動Logstash進程進行數(shù)據(jù)傳輸。模擬數(shù)據(jù)產(chǎn)生（1）用戶基本信息表導(dǎo)入選擇“基本查詢[+]”選項卡，可以查看Elasticsearch的用戶基本信息表的記錄模擬數(shù)據(jù)產(chǎn)生在Logstash安裝目錄的custom_config目錄下創(chuàng)建Logstash的配置文件mediamatch_userevent.conf，實現(xiàn)導(dǎo)入mediamatch_userevent.csv數(shù)據(jù)到Elasticsearch集群中。將mediamatch_userevent.csv文件上傳到server1（IP地址為3）的/opt/data/userevent目錄下，同時需要在sincedb_path目錄下創(chuàng)建userevent.sdp文件，用于保存讀取文件進度的記錄。啟動Logstash進程，將數(shù)據(jù)導(dǎo)入Elasticsearch集群的mediamatch_userevent索引。模擬數(shù)據(jù)產(chǎn)生（2）用戶狀態(tài)信息變更表導(dǎo)入Elasticsearch集群中用戶狀態(tài)信息變更表的記錄模擬數(shù)據(jù)產(chǎn)生在Logstash安裝目錄的custom_config目錄下創(chuàng)建Logstash的配置文件mm_billevents.conf，實現(xiàn)導(dǎo)入mmconsume_billevents.csv數(shù)據(jù)到Elasticsearch集群中。將mmconsume_billevents.csv文件上傳到server1（IP地址為192.168.111.73）的/opt/data/billevents目錄下，同時需要在sincedb_path目錄下創(chuàng)建billevents.sdp文件，用于保存讀取文件進度的記錄。啟動Logstash進程，將數(shù)據(jù)導(dǎo)入Elasticsearch集群的mmconsume_billevents索引。模擬數(shù)據(jù)產(chǎn)生（3）賬單信息表導(dǎo)入Elasticsearch集群中賬單信息表的記錄模擬數(shù)據(jù)產(chǎn)生在Logstash安裝目錄的custom_config目錄下創(chuàng)建Logstash的配置文件order_index.conf。將order_index_v3.csv文件上傳到server1（IP地址為3）的/opt/data/order_index目錄下，同時需要在sincedb_path目錄下創(chuàng)建order.sdp文件，用于保存讀取文件進度的記錄。啟動Logstash進程，將數(shù)據(jù)導(dǎo)入Elasticsearch集群的order_index集群索引。模擬數(shù)據(jù)產(chǎn)生（4）訂單信息表導(dǎo)入Elasticsearch集群中訂單信息表的記錄模擬數(shù)據(jù)產(chǎn)生在Logstash安裝目錄的custom_config目錄下創(chuàng)建Logstash的配置文件media1.conf，實現(xiàn)導(dǎo)入用戶收視行為信息數(shù)據(jù)到Elasticsearch集群中。用戶收視行為信息表的數(shù)據(jù)量比較大，可以選擇將原數(shù)據(jù)文件分成多個文件，這樣在多個節(jié)點上啟動Logstash進程，將數(shù)據(jù)導(dǎo)入Elasticsearch集群的效率會更高。配置文件只需要在input的file插件中修改對應(yīng)的數(shù)據(jù)路徑即可。啟動Logstash進程進行數(shù)據(jù)傳輸。模擬數(shù)據(jù)產(chǎn)生（2）用戶收視行為信息表導(dǎo)入Elasticsearch集群中用戶收視行為信息表的記錄模擬數(shù)據(jù)產(chǎn)生（1）模擬產(chǎn)生賬單數(shù)據(jù)數(shù)據(jù)產(chǎn)生規(guī)則因為賬單數(shù)據(jù)是每月1日生成的，賬單信息表中的year_month字段記錄了賬單時間，所以只需要每月更新year_month字段的值，得到時間不同的數(shù)據(jù)，并將其作為每月產(chǎn)生的新的數(shù)據(jù)即可。更改year_month字段的規(guī)則是先于每月1日計算當(dāng)前年月減原始賬單數(shù)據(jù)中最大的年月（2018-07）得到相差的月數(shù)month_delta，再將mmconsume_billevents數(shù)據(jù)中的year_month字段都加上month_delta，最后將修改后的數(shù)據(jù)更新到Elasticsearch集群中。模擬數(shù)據(jù)產(chǎn)生3．模擬產(chǎn)生數(shù)據(jù)具體步驟參考4.2.1小節(jié)數(shù)據(jù)從Elasticsearch傳輸?shù)紿ive中的方法，將Elasticsearch集群中的mmconsume_billevents數(shù)據(jù)同步到Hive的user_profile庫的mmconsume_billevents_id表中，模擬產(chǎn)生的賬單數(shù)據(jù)都是以Hive中的mmconsume_billevents表為基礎(chǔ)的。編寫mmconsume_billevents_1d.sh腳本，主要用于計算當(dāng)前時間與2018年7月相差的月數(shù)month_delta，即將原賬單數(shù)據(jù)的year_month字段數(shù)據(jù)加上month_delta，修改后的數(shù)據(jù)以覆蓋的模式保存在Hive的user_profile庫的mmconsume_billevents_1d表中，最后通過運行一個Spark程序?qū)ive中的mmconsume_billevents_1d表中的數(shù)據(jù)以覆蓋的模式保存到Elasticsearch中。模擬數(shù)據(jù)產(chǎn)生為了確保mmconsume_billevents_1d.sh腳本運行成功，需要先在腳本所在的節(jié)點上創(chuàng)建/root/qwm/mmconsume目錄，再將Hive數(shù)據(jù)導(dǎo)入Elasticsearch的代碼打包成user_profile_project-1.0.jar的JAR包，后續(xù)模擬產(chǎn)生訂單數(shù)據(jù)和用戶收視行為數(shù)據(jù)都需要使用這個JAR包。為了測試mmconsume_billevents_1d.sh腳本能否正確運行，在2018年8月28日設(shè)置了一個定時策略。在server1節(jié)點上執(zhí)行“crontab–e”，加入命令“01031*?*/root/mmconsume_billevents_1d.sh2018-07-015server3/root/qwm/user_profile_project-1.0.jar9200year_monthmmconsume_billevents_updatedoc”。在策略中設(shè)置的定時任務(wù)是在每月1日凌晨3點10分觸發(fā)的，這樣可以通過在2018年9月1日早上觀察任務(wù)的執(zhí)行結(jié)果來驗證編寫的腳本是否正確。模擬數(shù)據(jù)產(chǎn)生2018年9月1日查看的執(zhí)行結(jié)果模擬數(shù)據(jù)產(chǎn)生數(shù)據(jù)產(chǎn)生規(guī)則訂單數(shù)據(jù)是每天都有記錄產(chǎn)生的，因此需要每天模擬產(chǎn)生新的數(shù)據(jù)。選擇訂單信息表中optdate≥2018-01-0100:00:00的記錄，使用這些數(shù)據(jù)來模擬產(chǎn)生訂單數(shù)據(jù)，具體的規(guī)則如下。以2018-10-10作為基準(zhǔn)task_start_time，計算該日期與2018-01-01間隔的天數(shù)delta。計算當(dāng)前日期與task_start_time間隔的天數(shù)curr_delta。每天從訂單信息表（optdate≥2018-01-0100:00:00）中選擇滿足（optdate字段的值+delta）=（當(dāng)前日期-curr_delta）的數(shù)據(jù)，并計算當(dāng)前日期與optdate間隔的天數(shù)delta1，同時將orderdate、expdate、effdate字段的值分別加上delta1，再將修改后的數(shù)據(jù)增量更新到Elasticsearch集群中。模擬數(shù)據(jù)產(chǎn)生（2）模擬產(chǎn)生訂單數(shù)據(jù)具體步驟。參考4.2.1小節(jié)數(shù)據(jù)從Elasticsearch傳輸?shù)紿ive中的方法，將Elasticsearch集群中的order_index_v3數(shù)據(jù)同步到Hive的user_profile庫的order_index_v3表中，模擬產(chǎn)生的訂單數(shù)據(jù)是以Hive的order_index_v3表中的optdate字段≥2018-01-0100:00:00的數(shù)據(jù)為基礎(chǔ)的。編寫order_index_1d.sh腳本，每天從Hive的order_index_v3表中提取符合模擬產(chǎn)生訂單數(shù)據(jù)規(guī)則的數(shù)據(jù)，并按規(guī)則修改orderdate、expdate和effdate這3個字段的值，以覆蓋的模式保存到Hive的order_index_1d表中，最后通過Spark程序?qū)rder_index_1d表中的數(shù)據(jù)以增量的模式導(dǎo)入Elasticsearch。模擬數(shù)據(jù)產(chǎn)生為了測試order_index_1d.sh腳本能否正確執(zhí)行，需要設(shè)置一個臨時的定時任務(wù)來測試腳本。在server1節(jié)點上執(zhí)行“crontab–e”，加入命令“0102**?*/root/order_index_1d.sh2018-10-10create_order_index_1d.hql2018-01-01server3/root/qwm/user_profile_project-1.0.jar59200optdateorder_index_updatedoc”。定時任務(wù)是在2021年9月4日設(shè)置的，order_index_1d.sh腳本在2021年9月15日凌晨2點10分會被觸發(fā)執(zhí)行。2021年9月15日查看的執(zhí)行結(jié)果模擬數(shù)據(jù)產(chǎn)生數(shù)據(jù)產(chǎn)生規(guī)則用戶收視行為的數(shù)據(jù)每天都會產(chǎn)生，因此需要每天模擬產(chǎn)生新的數(shù)據(jù)。使用2018年5月—2018年7月這3個月的用戶收視行為數(shù)據(jù)作為基準(zhǔn)，根據(jù)如下規(guī)則模擬產(chǎn)生數(shù)據(jù)。選擇某個日期為task_start_time，如選擇2018-10-08作為task_start_time，計算該日期與用戶收視行為數(shù)據(jù)的最小日期（2018-05-02）間隔的天數(shù)，即delta=（2018-10-08）-（2018-05-02）。計算當(dāng)前日期與task_start_time間隔的天數(shù)curr_delta。每天從用戶收視行為數(shù)據(jù)中選擇滿足（origin_time字段的值+delta）=（當(dāng)前日期-curr_delta）條件的數(shù)據(jù)，同時將時間字段（origin_time、end_time）修改成當(dāng)前日期，并將修改后的數(shù)據(jù)更新到Elasticsearch集群中。模擬數(shù)據(jù)產(chǎn)生（3）模擬用戶收視行為數(shù)據(jù)具體步驟。參考4.2.1小節(jié)數(shù)據(jù)從Elasticsearch傳輸?shù)紿ive中的方法，將Elasticsearch集群中的用戶收視行為數(shù)據(jù)同步到Hive的user_profile庫的media_index_3m表中。編寫media_index_1d.sh腳本，從用戶收視行為數(shù)據(jù)中篩選出時間符合規(guī)則的數(shù)據(jù)，并且將origin_time字段和end_time字段的日期修改為當(dāng)前日期，將修改后的數(shù)據(jù)以覆蓋的模式保存到Hive的user_profile庫的media_1d表中，最后通過Spark程序?qū)edia_1d表中的數(shù)據(jù)以增量的模式更新到Elasticsearch集群中。模擬數(shù)據(jù)產(chǎn)生為了測試media_index_1d.sh腳本是否正確，需要設(shè)置定時策略來測試腳本。在server1節(jié)點上輸入“crontab–e”，加入命令“0101**?*/root/media_index_1d.sh2018-10-10create_media_1d.hql2018-05-02server3/root/qwm/user_profile_project-1.0.jar59200origin_timemedia_indexmedia”|。定時任務(wù)是在2018年10月15日設(shè)置的，media_index_1d.sh腳本會在2018年10月16日凌晨1點10分被觸發(fā)執(zhí)行，將數(shù)據(jù)增量更新到Elasticsearch的media_indexyyyyww索引中，其中，yyyyww代表年份與周數(shù)，即以周為單位存儲用戶收視行為數(shù)據(jù)。2018年10月16日查看的執(zhí)行結(jié)果模擬數(shù)據(jù)產(chǎn)生1數(shù)據(jù)存儲與傳輸目錄業(yè)務(wù)數(shù)據(jù)2基礎(chǔ)數(shù)據(jù)預(yù)處理3SparkSQL提供了esDF(resource:String,query:String,cfg:scala.collection.Map[String,String])方法讀取Elasticsearch數(shù)據(jù)。esDF方法中的參數(shù)解釋Elasticsearch數(shù)據(jù)傳輸?shù)紿ive參數(shù)解釋resourceElasticsearch數(shù)據(jù)的index和type，如mediamatch_usermsg/docquery查詢語句cfgElasticsearch集群的相關(guān)配置，如Elasticsearch節(jié)點IP地址、端口號等讀取mediamatch_usermsg數(shù)據(jù)，并選取phone_no為5143217的數(shù)據(jù)。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive以mediamatch_usermsg數(shù)據(jù)為例，介紹如何從Elasticsearch中讀取指定時間范圍內(nèi)的數(shù)據(jù)。先從Elasticsearch中讀取mediamatch_usermsg中的所有數(shù)據(jù)為DataFrame，再將DataFrame注冊成為臨時表。得到臨時表后，可以通過一條SQL語句從臨時表中過濾指定時間范圍的數(shù)據(jù)。自定義getBeforeTimeStr(timePattern:String,timeRangeValue:Int,timeRangePattern:String,startTime:String)方法，將getBeforeTime方法返回的Date類型轉(zhuǎn)換為String類型。從注冊的臨時表中選取當(dāng)前時間之前50年的數(shù)據(jù)，并將選中的數(shù)據(jù)保存到Hive的user_profile庫的mediamatch_usermsg中。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive定義getBeforeTime(timePattern:String,timeRangeValue:Int,timeRangePattern:String,startTime:String)方法，該方法中有4個參數(shù)。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive參數(shù)解釋timePattern指時間數(shù)據(jù)的格式，如“yyyy-MM-ddHH:mm:ss”timeRangeValue、timeRangePattern指時間跨度，如若為50年，則timeRangeValue為50，timeRangePattern為“Y”startTime指從哪個時間開始往前截取數(shù)據(jù)，如截取“2018-08-0100:00:00”之前的數(shù)據(jù)，則startTime為“2018-08-0100:00:00”media_index與其他4份數(shù)據(jù)不一樣，media_index是根據(jù)時間，將每周的數(shù)據(jù)保存到一張表（一個index）中。讀取media_index數(shù)據(jù)的方式也與mediamatch_usermsg等數(shù)據(jù)稍有不同。讀取media_index數(shù)據(jù)需要將指定時間范圍內(nèi)的所有與media_index相關(guān)的index/type找到，如讀取2018-08-01前10天的數(shù)據(jù)，需要匹配到media_index201830/media和media_index201831/media。為了能夠根據(jù)指定的時間范圍獲取media_index的所有index/type，需要自定義getRangeDays方法和getBeforeTimeTableNames方法來實現(xiàn)。Elasticsearch數(shù)據(jù)傳輸?shù)紿ivegetRangeDays方法用于獲取指定時間范圍內(nèi)的所有日期，如獲取2018-08-01前10天的日期，使用getRangeDays方法獲取的即2018-07-22至2018-07-31這10天的日期。getBeforeTimeTableNames方法有7個參數(shù)Elasticsearch數(shù)據(jù)傳輸?shù)紿ive參數(shù)解釋timePattern指時間數(shù)據(jù)的格式，如“yyyy-MM-ddHH:mm:ss”timeRangeValue、timeRangePattern指時間跨度，若為50年，則timeRangeValue為50，timeRangePattern為“Y”esIndexPre指的是index的前綴，如index為“media_index201831”中的“media_index”exIndexType指的是index后綴的類型，media_index每周會形成一個index，因此esIndexType為“yyyyww”esType指的是media_index數(shù)據(jù)的type，即“media”startTime指從哪個時間開始往前截取數(shù)據(jù)，如要截取“2018-08-0100:00:00”之前的數(shù)據(jù)，則startTime為“2018-08-0100:00:00”為了提升程序的健壯性，需要對getBeforeTimeTableNames方法得到的index/type列表進行進一步篩選，篩選出Elasticsearch數(shù)據(jù)中存在的index/type。為判斷index/type是否存在于Elasticsearch數(shù)據(jù)中并獲取到判斷結(jié)果，需要使用RestClient類與Elasticsearch集群進行通信，通過RestClient類中的typeExists(Stringindex,Stringtype)方法判斷index/type是否存在。然后通過列表的map操作，使用esDF方法獲取index/type的數(shù)據(jù)并將其轉(zhuǎn)換為DataFrame，得到一個包含多個DataFrame元素的列表，因為最后需要將獲取到的所有數(shù)據(jù)存儲到同一張Hive表中，所以可以先將列表中的多個DataFrame合并成一個DataFrame。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive因為后續(xù)需要定時將Elasticsearch數(shù)據(jù)傳輸?shù)紿ive中，為了方便任務(wù)的調(diào)度，所以對Elasticsearch數(shù)據(jù)傳輸?shù)紿ive中的代碼進行封裝。Elasticsearch中的數(shù)據(jù)有兩種類型：一種是一張表的數(shù)據(jù)，如mediamatch_usermsg；另一種是多張表的數(shù)據(jù)，如media_index。要將這兩種類型的數(shù)據(jù)傳輸?shù)紿ive中，需要封裝兩個類。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive針對將一張表的數(shù)據(jù)傳輸?shù)紿ive中的封裝類Elasticsearch2Hive，設(shè)置的參數(shù)如下。selectedCols：Elasticsearch資源同步的列字段名。timeColName：Elasticsearch資源時間列名稱。timeColPattern：資源時間列格式，如yyyyMMddHH:mm:ss。timeRangeUalue：Elasticsearch資源同步時間段值。timeRangeType：Elasticsearch資源同步時間段類型（Y|M|D。esTable：Elasticsearch資源名，即index/type。startTime：同步設(shè)置的某個時間之前的數(shù)據(jù)。Elasticsearch數(shù)據(jù)傳輸?shù)紿ive針對將多張表的數(shù)據(jù)傳輸?shù)紿ive中的封裝類ElasticsearchMulti2Hive，設(shè)置的參數(shù)如下。selectedCols：Elasticsearch資源同步的列字段名。tsColsName：Elasticsearch資源時間列名稱。tsColPattem：Elasticsearch資源時間列格式。timeRangeUalue：Elasticsearch資源同步時間段值。timeRangeType：Elasticsearch資源同步時間段類型（Y|M|D）。Elasticsearch數(shù)據(jù)傳輸?shù)紿iveElasticsearchMulti2Hive，設(shè)置的參數(shù)如下。esIndexType：Elasticsearch資源名index后綴類型（yyyyMMdd|yyyyMM|yyyyww），yyyyMMdd表示每天一個index，yyyyMM表示每月一個index，yyyyww表示每周一個index。esIndexPre：Elasticsearch資源名index前綴。esType：Elasticsearch資源名的類型，設(shè)置為“media”。startTime：同步設(shè)置的某個時間之前的數(shù)據(jù)。Elasticsearch數(shù)據(jù)傳輸?shù)紿iveSparkSQL實現(xiàn)用戶畫像得到的是DataFrame類型的數(shù)據(jù)，包含3個字段，分別為phone_no（用戶編號）、label（標(biāo)簽）、parent_label（父標(biāo)簽）。要想將DataFrame類型的數(shù)據(jù)保存到MySQL中，需要在MySQL數(shù)據(jù)庫中設(shè)計并創(chuàng)建一張標(biāo)簽表user_label，用于保存用戶畫像標(biāo)簽結(jié)果數(shù)據(jù)。用戶畫像標(biāo)簽結(jié)果保存到MySQLDataFrameWriter提供了jdbc方法將數(shù)據(jù)傳輸?shù)組ySQL中。jdbc方法的定義用戶畫像標(biāo)簽結(jié)果保存到MySQLjdbc方法中有3個參數(shù)，url參數(shù)指的是連接至MySQL數(shù)據(jù)庫的URL，如jdbc:mysql://5:3306/user_profile，table參數(shù)指的是要存放數(shù)據(jù)的表名，connectionProperties參數(shù)指的是連接數(shù)據(jù)庫的配置，如用戶名和密碼。為測試是否可以將DataFrame數(shù)據(jù)保存到MySQL中，計算得到品牌名稱的標(biāo)簽數(shù)據(jù)data。將data保存到MySQL中，設(shè)置url