與Pyspark

Question

並行化的HTTP請求我使用pyspark做一些數據轉換：類似如下：

df_systems_tree_users = sqlContext.read.format("jdbc") \ 
    .option("dbtable", 
      "(select ID as SYSTEMUID,M_EXTERNAL_ID,metric,DATATRANSMISSIONFREQUENCY,MODEL,BRAND,BUILDING FROM SYSTEM INNER JOIN SENSOR ON SYSTEM.ID=SENSOR.SYSTEMID WHERE LPWANOPERATOR='Objenious' AND M_EXTERNAL_ID!='None')") \ 
    .option("url", "jdbc:phoenix:master1:2181:/hbase-unsecure") \ 
    .option("driver", "org.apache.phoenix.jdbc.PhoenixDriver") \ 
    .load() 
objRDD = df_systems_tree_users.rdd.map(lambda x: getStatesAndUplink(x)) 

getStatesAndUplink方法使用請求python庫來執行HTTP GET請求外部API。

我在4個執行器上運行4個執行器，每個執行器有4個內核，但運行30分鐘需要很多時間。

我的問題是如何優化我的代碼以高效的方式並行化我的http請求？

Answer 1

正如documentation提到的，你必須指定4個參數：

• partitionColumn，
• lowerBound，
• upperBound，
• numPartitions

只有這些選項星火會並行讀入其他情況下，它將在一個線程中完成。

編輯：存在Phoenix Spark plugin。它將並行讀取而不指定這4個參數。

編輯2：如果數據不平衡，則方法getStatesAndUplink可能會限制外部服務。它可能只是「卡住」在一個節點上的處理