性能問題與pyspark udf調用rest api函數

[sensanjoy]

你好,

我麵臨一些性能問題的pyspark udf post數據的REST API函數(使用宇宙數據庫後端存儲的數據)。

請找到下麵的細節:

#火花dataframe (df)包含關於數據30-40k附近。

#我是使用python udf張貼在rest api函數:

例如final_df = df。withColumn(“地位”,save_data (A, B, C))

# udf功能:

@udf (returnType = IntegerType ())

def save_data (A, B, C):

post_data =列表()

post_data.append ({

A:,

B: B,

“C”: C,

})

retry_strategy =重試(

總= 5,

status_forcelist = [400、500、502、503、504),

method_whitelist =(“文章”),

backoff_factor = 0.1

)

適配器= HTTPAdapter (max_retries = retry_strategy)

s = requests.Session ()

年代。山(“https://”,適配器)

年代。山(“http://”,適配器)

年代。keep_alive = False

試一試:

響應= s.post (

url = rest_api_url,

頭={“授權”:“持票人”+ api_token,“內容類型”:“application / json "},

data = json.dumps (post_data)

)

返回response.status_code

除了:

響應= requests.post (

url = rest_api_url,

頭={“授權”:“持票人”+ api_token,“內容類型”:“application / json "},

data = json.dumps (post_data)

)

返回response.status_code

#問題:磚工作被絞死的無限次rest api調用(save_data()),從未成功。

#當檢查從API,它顯示了服務接觸最大資源利用率(100%)。

對我來說它就像python udf是一次發送大量數據淹沒了api服務在某種程度上的時間和它停止響應。

我們將是最好的辦法可以解決這個問題?

我們應該把dataframe分成多個塊和一個一個發送還是

把它轉換成熊貓df然後發送逐行(可能是慢)

請建議。

[匿名 · Accepted solution]

@Sanjoy森:

看起來UDF函數使得同步HTTP請求dataframe中的每一行的REST API,這可能會導致性能問題在處理大量的數據。

為了提高性能,您可以考慮以下方法:

1. 批量數據:每一行發送到REST API,而是可以批處理數據和發送多個行。這將減少HTTP請求的數量,提高整體性能。您可以使用foreachPartition火花成批發送數據的API。
2. 使用異步HTTP客戶端:您可以使用異步HTTP客戶端像aiohttp或httpx UDF函數異步發送數據。這將允許您發送多個HTTP請求並行,提高整體性能。
3. 使用一個分布式係統:如果要處理的數據太大在同一台計算機上,您可以考慮使用分布式係統(比如Apache卡夫卡或Apache火花流媒體實時處理數據並將其發送到REST API。
4. 使用專用數據處理框架:你可以用一個專門的數據處理框架(如Apache梁或Apache Flink處理數據並將其發送到REST API。這些框架被設計為處理大量數據和攝入可以處理數據,處理,和交付。

一般來說,從一個UDF可以發送數據逐行效率低下,導致性能問題,尤其是在處理大型數據集。最好使用批處理和異步HTTP請求來提高性能。

[sensanjoy]

@Suteja卡努裏人感謝你的輸入和真的有意義你提到批量數據發送請求比逐行提供的API支持批量插入。我也會考慮異步http請求。

[匿名]

@Sanjoy森:這是美妙的!

[Vidula_Khanna]

嗨森@Sanjoy

謝謝你的問題!幫助你更好的,請花一些時間來檢查答案,讓我知道它是否最適合您的需要。

請幫助我們選擇最好的解決方案通過點擊“選擇最佳”如果它。

您的反饋將幫助我們確保我們提供最好的服務給你。謝謝你！