PyODPS DataFrame 的代碼在哪裏跑

在使用 PyODPS DataFrame 編寫數據應用時，儘管編寫的是同一個腳本文件，但其中的代碼會在不一樣位置執行，這可能致使一些沒法預期的問題，本文介紹當出現相關問題時，如何肯定代碼在何處執行，以及提供部分場景下解決問題的方法。

概述

假定咱們要執行下面的代碼：

from odps import ODPS, options
import numpy as np

o = ODPS(access_id, access_key, project, endpoint)
df = o.get_table('pyodps_iris').to_df()

coeffs = [0.1, 0.2, 0.4]

def handle(v):
    import numpy as np
    return float(np.cosh(v)) * sum(coeffs)

options.df.supersede_libraries = True
val = df.sepal_length.map(handle).sum().execute(libraries=['numpy.zip', 'other.zip'])
print(np.sinh(val))

在開始分析以前，首先須要指出的是，PyODPS 是一個 Python 包而非 Python Implementation，PyODPS 的運行環境均爲未經修改的 Python，於是並不會出現與正常 Python 解釋器不一致的行爲。亦即，你所寫的每一條語句不會有與標準 Python 語句不一樣的行爲，例如自動變成分佈式代碼，等等。

下面解釋該代碼的執行過程。

上圖是執行上述代碼時可能涉及的系統。代碼自己執行的位置在圖中用紫色表示，這些系統都位於 MaxCompute 外部，爲了方便表述，下文稱爲「本地」。在本地執行的代碼包括 handle 函數以外的部分（注意 handle 傳入 map 時僅傳入了函數自己而並未執行）。於是，這些代碼在執行時，行爲與普通 Python code 的執行行爲相似，import 第三方包時，引用的是本地的包。於是，上面的代碼中，libraries=['numpy.zip', 'other.zip']引用的other.zip由於並無在本地安裝，於是若是代碼中有諸如 import other 這樣的語句，會致使執行報錯。即使 other.zip 已被上傳到 MaxCompute 資源也是如此，由於本地根本沒有這個包。理論上，本地代碼若是不涉及 PyODPS 包，則與 PyODPS 無關，用戶須要自行排查。

對於 handle 函數，狀況發生了變化。handle 函數傳入 map 方法時，若是使用的後端是 MaxCompute 後端，會先被 cloudpickle 模塊提取閉包和字節碼，此後 PyODPS DataFrame 會使用閉包和字節碼生成一個 Python UDF，提交到 MaxCompute。最後，做業以 SQL 的形式在 MaxCompute 執行時，會調用這個 Python UDF，其中的字節碼和閉包內容會被 unpickle，此後在 MaxCompute Executor 執行。因而可知，在上述代碼中，

1. 在 handle 函數體中的代碼都不會在本地執行，而會在 MaxCompute Executor 中執行；
2. handle 函數體中沒法引用本地安裝的包，只有在 MaxCompute Executor 中存在的包纔有效；
3. 上傳的第三方包必須可以在 MaxCompute Executor 中的 Python 版本（目前爲 Python 2.7，UCS2）中調用；
4. handle 函數體中修改引用的外部變量（上述代碼中的 coeffs）不會致使本地的 coeffs 值被修改；
5. 若是在 handle 中引用在 handle 外 import 的包，在 handle 中調用可能會報錯，由於在不一樣環境中，包的結構可能不一樣，而 cloudpickle 會將本地包的引用帶到 MaxCompute Executor，致使報錯，於是建議 import 在 handle 中進行；
6. 因爲使用 cloudpickle，若是在 handle 中調用了其餘文件中的代碼，該文件所在的包必須存在於 MaxCompute Executor 中。若是你不想使用第三方包的形式解決該問題，請將全部引用的我的代碼放在同一個文件中。

上述對 handle 函數的解釋對於自定義聚合、apply 和 map_reduce 中調用的自定義方法 / Agg 類均適用。若是使用的後端是 Pandas 後端，則全部代碼都會在本地運行，於是本地也須要安裝相關的包。但鑑於 Pandas 後端調試完畢後一般會轉移到 MaxCompute 運行，建議在本地裝包的同時，參照 MaxCompute 後端的慣例進行開發。

使用第三方包

1. 我的電腦 / 自有服務器在本地使用第三方包 / 其餘文件中的代碼

在相應的 Python 版本上安裝便可。

1. DataWorks 中本地使用其餘文件中的代碼

該部分功能由 DataWorks 提供，請參考 DataWorks 文檔。

1. map / apply / map_reduce / 自定義聚合中使用第三方包 / 其餘文件中的代碼

參考 https://yq.aliyun.com/articles/591508 。須要補充的是，在 DataWorks 上上傳資源後，須要點擊「提交」確保資源被正確上傳到 MaxCompute。若是須要使用本身的 Numpy 版本，在上傳正確版本的 wheel 包的同時，須要配置 odps.df.supersede_libraries = True，同時確保你上傳的 numpy 包名位於 libraries 的最前面，若是指定了 options.df.libraries，則 numpy 包名須要位於 options.df.libraries 的最前面。

引用其餘 MaxCompute 表中的數據

1. 我的電腦 / 自有服務器在本地訪問 MaxCompute 表

若是 Endpoint 能夠鏈接，使用 PyODPS / DataFrame 訪問。

1. map / apply / map_reduce / 自定義聚合中訪問其餘 MaxCompute 表

MaxCompute Executor 中一般不支持訪問 Endpoint / Tunnel Endpoint，其上也沒有 PyODPS 包可用，於是不能直接使用 ODPS 入口對象或者 PyODPS DataFrame，也不能從自定義函數外部傳入這些對象。若是表的數據量不大，建議將 DataFrame 做爲資源傳入（見 https://pyodps.readthedocs.io/zh_CN/latest/df-element.html#function-resource ）。若是數據量較大，建議改寫成 join。

訪問其餘服務

1. 我的電腦 / 自有服務器在本地訪問其餘服務

保證本身的環境中能夠正常訪問相關服務，生產服務器能夠聯繫 PE。

1. DataWorks 上的本地代碼中訪問其餘服務

請諮詢 DataWorks。

1. map / apply / map_reduce / 自定義聚合中訪問其餘服務

參考 https://yq.aliyun.com/articles/591508 啓用 Isolation，若是仍然遇到網絡報錯，請聯繫 MaxCompute 用戶羣，找 PE 幫忙解決。

 

本文做者：繼盛

原文連接

本文爲雲棲社區原創內容，未經容許不得轉載。