今天就跟大家聊聊有關(guān)PyODPS DataFrame 處理笛卡爾積的幾種方式分別是什么，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

成都創(chuàng)新互聯(lián)是少有的成都做網(wǎng)站、網(wǎng)站制作、營(yíng)銷型企業(yè)網(wǎng)站、微信小程序定制開(kāi)發(fā)、手機(jī)APP,開(kāi)發(fā)、制作、設(shè)計(jì)、外鏈、推廣優(yōu)化一站式服務(wù)網(wǎng)絡(luò)公司,自2013年起,堅(jiān)持透明化,價(jià)格低,無(wú)套路經(jīng)營(yíng)理念。讓網(wǎng)頁(yè)驚喜每一位訪客多年來(lái)深受用戶好評(píng)

PyODPS 提供了 DataFrame API 來(lái)用類似 pandas 的接口進(jìn)行大規(guī)模數(shù)據(jù)分析以及預(yù)處理，本文主要介紹如何使用 PyODPS 執(zhí)行笛卡爾積的操作。

笛卡爾積最常出現(xiàn)的場(chǎng)景是兩兩之間需要比較或者運(yùn)算。以計(jì)算地理位置距離為例，假設(shè)大表 Coordinates1 存儲(chǔ)目標(biāo)點(diǎn)經(jīng)緯度坐標(biāo)，共有 M 行數(shù)據(jù)，小表 Coordinates2 存儲(chǔ)出發(fā)點(diǎn)經(jīng)緯度坐標(biāo)，共有 N 行數(shù)據(jù)，現(xiàn)在需要計(jì)算所有離目標(biāo)點(diǎn)最近的出發(fā)點(diǎn)坐標(biāo)。對(duì)于一個(gè)目標(biāo)點(diǎn)來(lái)說(shuō)，我們需要計(jì)算所有的出發(fā)點(diǎn)到目標(biāo)點(diǎn)的距離，然后找到最小距離，所以整個(gè)中間過(guò)程需要產(chǎn)生 M * N 條數(shù)據(jù)，也就是一個(gè)笛卡爾積問(wèn)題。

haversine 公式

首先簡(jiǎn)單介紹一下背景知識(shí)，已知兩個(gè)地理位置的坐標(biāo)點(diǎn)的經(jīng)緯度，求解兩點(diǎn)之間的距離可以使用 haversine 公式，使用 Python 的表達(dá)如下：

def  haversine(lat1,  lon1,  lat2,  lon2):
        #  lat1,  lon1  為位置  1  的經(jīng)緯度坐標(biāo)
        #  lat2,  lon2  為位置  2  的經(jīng)緯度坐標(biāo)
        import  numpy  as  np

        dlon  =  np.radians(lon2  -  lon1)
        dlat  =  np.radians(lat2  -  lat1)
        a  =  np.sin(  dlat  /2  )  **2  +  np.cos(np.radians(lat1))  *  np.cos(np.radians(lat2))  *  np.sin(  dlon  /2  )  **2
        c  =  2  *  np.arcsin(np.sqrt(a))
        r  =  6371  #  地球平均半徑，單位為公里
        return  c  *  r

MapJoin

目前最推薦的方法就是使用 mapjoin，PyODPS 中使用 mapjoin 的方式十分簡(jiǎn)單，只需要兩個(gè) dataframe join 時(shí)指定 mapjoin=True，執(zhí)行時(shí)會(huì)對(duì)右表做 mapjoin 操作。

In  [3]:  df1  =  o.get_table('coordinates1').to_df()                                                                                                                                                                                        

In  [4]:  df2  =  o.get_table('coordinates2').to_df()                                                                                                                                                                                        

In  [5]:  df3  =  df1.join(df2,  mapjoin=True)                                                                                                                                                                                                        

In  [6]:  df1.schema                                                                                                                                                                                                                                                      
Out[6]:  
odps.Schema  {
    latitude                    float64              
    longitude                  float64              
    id                                string                
}

In  [7]:  df2.schema                                                                                                                                                                                                                                                      
Out[7]:  
odps.Schema  {
    latitude                    float64              
    longitude                  float64              
    id                                string                
}

In  [8]:  df3.schema                                                                                                                                                                                                                                                      
Out[8]:  
odps.Schema  {
    latitude_x                        float64              
    longitude_x                      float64              
    id_x                                    string                
    latitude_y                        float64              
    longitude_y                      float64              
    id_y                                    string                
}

可以看到在執(zhí)行 join 時(shí)默認(rèn)會(huì)將重名列加上 _x 和 _y 后綴，可通過(guò)在 suffixes 參數(shù)中傳入一個(gè)二元 tuple 來(lái)自定義后綴，當(dāng)有了 join 之后的表后，通過(guò) PyODPS 中 DataFrame 的自建函數(shù)就可以計(jì)算出距離，十分簡(jiǎn)潔明了，并且效率很高。

In  [9]:  r  =  6371  
      ...:  dis1  =  (df3.latitude_y  -  df3.latitude_x).radians()  
      ...:  dis2  =  (df3.longitude_y  -  df3.longitude_x).radians()  
      ...:  a  =  (dis1  /  2).sin()  **  2  +  df3.latitude_x.radians().cos()  *  df3.latitude_y.radians().cos()  *  (dis2  /  2).sin()  **  2  
      ...:  df3['dis']  =  2  *  a.sqrt().arcsin()  *  r                                                                                                                                                                                              
                                                                                                                                                                                                        
In [12]: df3.head(10)                                                                                                                        
Out[12]: 
    latitude_x  longitude_x id_x  latitude_y   longitude_y id_y       dis
0   76.252432    59.628253    0   84.045210     6.517522    0  1246.864981
1   76.252432    59.628253    0   59.061796     0.794939    1  2925.953147
2   76.252432    59.628253    0   42.368304    30.119837    2  4020.604942
3   76.252432    59.628253    0   81.290936    51.682749    3   584.779748
4   76.252432    59.628253    0   34.665222   147.167070    4  6213.944942
5   76.252432    59.628253    0   58.058854   165.471565    5  4205.219179
6   76.252432    59.628253    0   79.150677    58.661890    6   323.070785
7   76.252432    59.628253    0   72.622352   123.195778    7  1839.380760
8   76.252432    59.628253    0   80.063614   138.845193    8  1703.782421
9   76.252432    59.628253    0   36.231584    90.774527    9  4717.284949

In [13]: df1.count()                                                                                                                         
Out[13]: 2000

In [14]: df2.count()                                                                                                                         
Out[14]: 100

In [15]: df3.count()                                                                                                                         
Out[15]: 200000

df3 已經(jīng)是有 M * N 條數(shù)據(jù)了，接下來(lái)如果需要知道最小距離，直接對(duì) df3 調(diào)用 groupby 接上 min 聚合函數(shù)就可以得到每個(gè)目標(biāo)點(diǎn)的最小距離。

In [16]: df3.groupby('id_x').dis.min().head(10)                                                                                              
Out[16]: 
       dis_min
0   323.070785
1    64.755493
2  1249.283169
3   309.818288
4  1790.484748
5   385.107739
6   498.816157
7   615.987467
8   437.765432
9   272.589621

DataFrame 自定義函數(shù)

如果我們需要知道對(duì)應(yīng)最小距離的點(diǎn)的城市，也就是表中對(duì)應(yīng)的 id ，可以在 mapjoin 之后調(diào)用 MapReduce，不過(guò)我們還有另一種方式是使用 DataFrame 的 apply 方法。要對(duì)一行數(shù)據(jù)使用自定義函數(shù)，可以使用 apply 方法，axis 參數(shù)必須為 1，表示在行上操作。

表資源

要注意 apply 是在服務(wù)端執(zhí)行的 UDF，所以不能在函數(shù)內(nèi)使用類似于df=o.get_table('table_name').to_df() 的表達(dá)式去獲得表數(shù)據(jù)，具體原理可以參考PyODPS DataFrame 的代碼在哪里跑。以本文中的情況為例，要想將表 1 與表 2 中所有的記錄計(jì)算，那么需要將表 2 作為一個(gè)資源表，然后在自定義中引用該表資源。PyODPS 中使用表資源也十分方便，只需要將一個(gè) collection 傳入 resources 參數(shù)即可。collection 是個(gè)可迭代對(duì)象，不是一個(gè) DataFrame 對(duì)象，不可以直接調(diào)用 DataFrame 的接口，每個(gè)迭代值是一個(gè) namedtuple，可以通過(guò)字段名或者偏移來(lái)取對(duì)應(yīng)的值。

## use dataframe udf

df1 = o.get_table('coordinates1').to_df()
df2 = o.get_table('coordinates2').to_df()

def func(collections):
    import pandas as pd
    
    collection = collections[0]
    
    ids = []
    latitudes = []
    longitudes = []
    for r in collection:
        ids.append(r.id)
        latitudes.append(r.latitude)
        longitudes.append(r.longitude)

    df = pd.DataFrame({'id': ids, 'latitude':latitudes, 'longitude':longitudes})
    def h(x):        
        df['dis'] = haversine(x.latitude, x.longitude, df.latitude, df.longitude)
        return df.iloc[df['dis'].idxmin()]['id']
    return h

df1[df1.id, df1.apply(func, resources=[df2], axis=1, reduce=True, types='string').rename('min_id')].execute(
    libraries=['pandas.zip', 'python-dateutil.zip', 'pytz.zip', 'six.tar.gz'])

在自定義函數(shù)中，將表資源通過(guò)循環(huán)讀成 pandas DataFrame，利用 pandas 的 loc 可以很方便的找到最小值對(duì)應(yīng)的行，從而得到距離最近的出發(fā)點(diǎn) id。另外，如果在自定義函數(shù)中需要使用到三方包（例如本例中的 pandas）可以參考這篇文章。

全局變量

當(dāng)小表的數(shù)據(jù)量十分小的時(shí)候，我們甚至可以將小表數(shù)據(jù)作為全局變量在自定義函數(shù)中使用。

df1 = o.get_table('coordinates1').to_df()
df2 = o.get_table('coordinates2').to_df()
df = df2.to_pandas()

def func(x):
    df['dis'] = haversine(x.latitude, x.longitude, df.latitude, df.longitude)
    return df.iloc[df['dis'].idxmin()]['id']

df1[df1.id, df1.apply(func, axis=1, reduce=True, types='string').rename('min_id')].execute(
    libraries=['pandas.zip', 'python-dateutil.zip', 'pytz.zip', 'six.tar.gz'])

在上傳函數(shù)的時(shí)候，會(huì)將函數(shù)內(nèi)使用到的全局變量（上面代碼中的 df） pickle 到 UDF 中。但是注意這種方式使用場(chǎng)景很局限，因?yàn)?ODPS 的上傳的文件資源大小是有限制的，所以數(shù)據(jù)量太大會(huì)導(dǎo)致 UDF 生成的資源太大從而無(wú)法上傳，而且這種方式最好保證三方包的客戶端與服務(wù)端的版本一致，否則很有可能出現(xiàn)序列化的問(wèn)題，所以建議只在數(shù)據(jù)量非常小的時(shí)候使用。

使用 PyODPS 解決笛卡爾積的問(wèn)題主要分為兩種方式，一種是 mapjoin，比較直觀，性能好，一般能用 mapjoin 解決的我們都推薦使用 mapjoin，并且最好使用內(nèi)建函數(shù)計(jì)算，能到達(dá)最高的效率，但是它不夠靈活。另一種是使用 DataFrame 自定義函數(shù)，比較靈活，性能相對(duì)差一點(diǎn)（可以使用 pandas 或者 numpy 獲得性能上的提升），通過(guò)使用表資源，將小表作為表資源傳入 DataFrame 自定義函數(shù)中，從而完成笛卡爾積的操作。

看完上述內(nèi)容，你們對(duì)PyODPS DataFrame 處理笛卡爾積的幾種方式分別是什么有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝大家的支持。

文章名稱：PyODPSDataFrame處理笛卡爾積的幾種方式分別是什么
URL分享：http://chinadenli.net/article10/ipsddo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供用戶體驗(yàn)、微信公眾號(hào)、建站公司、定制開(kāi)發(fā)、網(wǎng)站收錄、微信小程序

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)