大資料-一站式内容畫像資料處理（pyodps+odps）

文章目錄

• 背景
• 資料流
• 原理
• 實戰示範
• • 資料內建
  • 資料開發（odps sql）
  • 資料開發（pyodps）
• pyodps實戰示範
• • 在dataworks中使用
  • 在本機使用
  • 在dsw的hello world
  • 在dsw中進行埋點分析（時間序列分析）
• 工程選擇
• • 工程中存在以下問題
  • 工程選擇
• 評價

背景

公司的pro環境并沒有搭建自己存儲環境，而是直接使用阿裡雲的rds，那麼數倉的建設也就直接使用了阿裡雲的dataworks（也就是原odps，也有maxcompute）。

資料存儲的限制，需要使用pyodps的sdk才能進行更好的管理、分析、處理（機器學習、深度學習模型訓練等）。

本wiki将闡述使用pyodps的優勢。

資料流

概念解析：

• 内容源
  • 爬蟲系統，scrapy實作，爬取文章、音視訊、标簽、評論輿情資訊、商品資訊等
  • 大v背景，大v釋出文章等
  • 營運背景，運作标簽系統
  • 使用者行為埋點，收集使用者操作app的行為日志，通過sls進行日志收集，在數倉進行處理，生成埋點資料表
• 資料內建，也可以被簡單的了解為資料同步，阿裡雲使用datax實作
• 數倉，也就是odps（阿裡雲參考hive實作）
• 資料處理
  • 分詞，jieba分詞
  • word2vec，文本向量化，也可以稱之為詞嵌入（embedding）
  • 主題聚類，lda、kmeans等聚類
  • 特征抽取，知識圖譜、關鍵詞提取等
  • tfidf、textrank，一些非監督的文本向量化手段
  • 模式分類，即給定一定的樣本後，訓練相應的分類模型，進行訓練和預測
• rds，阿裡雲的資料庫；nosql，包括redis、hbase、MongoDB、elasticsearch等
• 腳本同步，python、go、java實作的一些簡單的同步腳本
• 資料應用，資料被整合和處理完畢之後，可以提供給不同的業務系統進行使用
  • 推薦系統，包括内容資訊流、電商推薦、廣告推薦
  • 其他業務
  • 資料挖掘，優質文章挖掘、人群關注的關鍵詞挖掘、輿情挖掘等

可以看到，資料最終是通過數倉進行資料內建的，那麼資料并不在本地，需要通過pyodps的sdk進行操作。

原理

以下為簡易說明圖

阿裡雲的odps，和hadoop的hive如出一轍，生成的hive表可以直接使用hive sql進行操作，最終的原理就是map reduce的過程，把生成的一個個任務分發到了不同的叢集機器上執行（具體原理可以研習一下hadoop的map reduce）。

odps的map reduce應該是完全類似于hive的，spark是另外一種map reduce的架構，隻不過是把硬碟上的操作挪移到了内容，加快了運算速度。

pyodps、pai、odps sql等都是對map reduce和spark進行了一次封裝，其中sql适合于表關聯的查詢統計等，而pyodps則摻雜了python操作，把表轉換成了類似于pandas中的DataFrame，可以完成複雜的程式設計類操作。機器學習pai，也可以在pyodps中進行調用，從何進行機器學習的相關功能。

實戰示範

資料內建

又名

抽數

。直接在dataworks中進行配置即可，這裡不加以贅述。

資料開發（odps sql）

也可以認為是hive sql（即hql），和普通的sql差別并不大，其中多了表分區的概念，同時也減少了一些概念（比如說索引等）。這時候産生了一個問題：“這個和mysql沒啥差別呀”。

然而并非如此，業務系統使用的是操作型資料庫，例如mysql；而odps（hive）資料分析型資料庫，隻做資料分析（最好是t+1的資料）。

資料開發（pyodps）

類似的開源産品有pyhive（但是據說有坑）。使用sql進行聯表的統計無疑是最佳選擇，但如果需要對資料進行jieba分詞呢？或者需要對已有資料進行lda主題模型分析、相似度分析、設定是做更多的複雜邏輯，使用sql來操作就顯得捉襟見肘了。

而python則不同，可以直接使用jieba子產品進行分詞，sklearn中同時也存在豐富的機器學習、nlp、資料處理的方法，另外還可以進行直接使用tensorflow進行深度學習，一切都顯得那麼美好。

pyodps實戰示範

參考文檔：

https://pyodps.readthedocs.io/zh_CN/latest/

https://help.aliyun.com/document_detail/121718.html?spm=a2c4g.11186623.6.783.5eeb153a3mGCDM

https://help.aliyun.com/document_detail/137514.html?spm=5176.10695662.1996646101.searchclickresult.52c4438fy1aKdO

在dataworks中使用

直接在對應的目錄，右鍵就可以直接建立pyodps腳本，下面簡單進行一下hello world

下面我們來實作一個jieba分詞的腳本

from odps import ODPS, options
from odps.df import DataFrame
from odps.df import output
import pandas as pd
import datetime
# 設定了這個标志，在print的時候就直接列印結果
options.interactive = True
# 導入jieba的包
options.df.libraries = ['jieba-master.zip']

# 實作jieba分詞函數
@output(['content_id', 'content_type', 'title', 'title_seg'], ['string', 'int', 'string', 'string'])
def seg(resources):
    stop_words = set([r[0] for r in resources[0]])
    def func(row):
        import jieba
        import sys
        import re
        pattern = u"^[a-zA-Z0-9\u4e00-\u9fa5]{0,}$"
        reload(sys)
        sys.setdefaultencoding('utf-8')
        word_list = jieba.lcut(row.title)
        word_list = [word.encode('utf-8') for word in word_list if word.encode('utf-8') not in stop_words and re.match(pattern=pattern, string=word)]
        return row.article_id, 1, row.title, ' '.join(word_list)
    return func

# 擷取停用詞（部落客的資源下載下傳裡面有停用詞的txt檔案）
def get_stop_words():
    resource = o.get_resource('stop_words.txt')
    stop_words = list()
    with resource.open('r') as fp:
        lines = fp.readlines()
        for line in lines:
            stop_words.append(line.strip().encode('utf-8'))
    stop_words = DataFrame(pd.DataFrame({'stop_words': stop_words}))
    return stop_words

# 計算當天時間
today = datetime.date.today()
oneday = datetime.timedelta(days=1)
yesterday = today - oneday
# 擷取源資料表
t = o.get_table('ztjy_dim.dim_flow_article_a').get_partition('ds={}'.format(str(yesterday)))
# 轉成df，這裡也可以直接調用to_df
article_df = DataFrame(t)
df = article_df[['article_id','title','is_online','is_delete']]
# 過濾上線的和未删除的
df = df.filter(df.is_online == 1, df.is_delete == 0)
# 分發給叢集執行
df = df.apply(seg, axis=1, resources=[get_stop_words()])
# print df.execute()
# 對表進行持久化
df = df.persist('ads_pai_content_portrait_title_seg_d', partition='ds={}'.format(str(yesterday)) , drop_partition=True, create_partition=True)
           

在本機使用

通過使用阿裡雲的ak sk，在本機也可以進行相關調試。其他的都類似，隻不過在建立odps執行個體的時候，需要添加手動添加ak sk。

from odps import ODPS

o = ODPS('**your-access-id**', '**your-secret-access-key**', '**your-default-project**',
            endpoint='**your-end-point**')
           

本機操作有一點是非常不友善的，資料進行調試的時候，要進行長時間的下載下傳。資料本身是在阿裡雲上，而資料在展示的時候，必然需要進行大量的資料下載下傳，導緻調試極為不便，一個網絡不好，就會斷了。。

在dsw的hello world

通過機器學習pai的dsw notebook環境進行pyodps開發。此環境說白了就是一台阿裡雲的ecs機器，但是有良好的封裝（anaconda）和可視化。

是不是感覺親切很多，這其實就是python的jupyter notebook。

來檢視一下之前的分詞表

還可以進行資料分析

資料可視化

可見在dsw上進行開發，友善快捷的多，開發完畢之後，可以直接把代碼上上到dataworks上進行排程；也可以把代碼上到自己的ecs，通過xxljob進行排程。

在dsw中進行埋點分析（時間序列分析）

加載表，然後看一下埋點上報類型的分布

使用barh圖檢視一下比例

過濾feed_show（需要的事件類型）

使用data_range進行時間序列分割，然後進行value_count，檢視每個時間段的數量。通過時間段的資料，可以反映出，使用者在發現資訊流中的行為時間段

工程選擇

工程中存在以下問題

• dsw用于做pyodps開發，進行DataFrame的運算是沒問題的，但如果進行本地運算，本地資源就會成為瓶頸
• 模型的運算大多需要在本地實作（特别是sklearn提供的模型），例如tfidf、lda等，隻有在進行預測的時候，才需要在odps的資源組中進行（也就是在map reduce的叢集中）
• pyodps在dataworks中的排程，不能在本地運算模型，同樣存在資源限制的問題

工程選擇

• 最好是阿裡雲ecs，作為pyodps的載體
• 在ecs上安裝jupyterhub，友善開發人員合作開發
• 通過xxl排程的方式，每天生成模型檔案，并且進行模型檔案上傳

評價

• 在dsw中，同時也可以建立pytoch、tensorflow的開發環境，是以odps+深度學習的模式是完全可以搞
• 另外，大的訓練集可以直接放到oss上，在進行訓練的時候也可以友善使用
• 資料完全集中在odps，不用再去通路業務庫，資料的安全性得到了保障，初步達到了一站式的開發