我使用Py函數讀取一些數據從一個端點,寫成一個CSV文件到Azure BLOB位置。
我得到端點需要2查詢參數,param1 param2。所以一開始,我有一個dataframe paramDf兩列param1和param2。
param1 param2 12 25 45 95模式:paramDF: pyspark.sql.dataframe。DataFrame param1:字符串param2:字符串
我現在寫一個函數如下:
def executeRestApi (w): dlist =[]嚐試:響應= requests.get (DataUrl。形式at(token=TOKEN, oid=w.param1,wid=w.param2)) if(response.status_code==200): metrics=response.json()['data']['metrics'] dic={} dic['metric1'] = metrics['metric1'] dic['metric2'] = metrics['metric2'] dlist.append(dic) pandas.DataFrame(dlist).to_csv("../../dbfs/mnt/raw/Important/MetricData/listofmetrics_{}_{}.csv".format(param1,param2),header=True,index=False) return "Success" except Exception as e: return "Failure"
最後,調用方法:
paramDf.foreach (executeRestApi)
因此,從理論上講,函數executeRestApi必須敵人dataframe中的每一行執行,在函數內,我提取所需的數據並將它寫入ADLS位置為csv文件。
所有作品好,除了文件從來沒有當我寫一個多節點集群上執行foreach命令。
然而同樣的操作單個節點集群上運行良好。我無法找出這兩種方法之間的區別。
可能我做錯了什麼呢?
。