有可能使用多進程或線程提交多個查詢並行數據庫從磚嗎?

[vanepet]

我們正在努力提高我們的整體運行時通過運行並行查詢使用多進程或線程。我看到的是,當運行這段代碼的功能是運行在一個單獨的進程並不返回dataFrame與任何信息。

示例代碼

def GetData (job_context group_numbers): sqlquery = " " SELECT * FROM偵察。v_ods_payment group_number在({0})”““.format (group_numbers) payments_df = job_context.create_dataframe_from_staging (sqlquery)打印(payments_df) #顯示(payments_df)返回payments_df def DoWork (): job_list =[”支付。在job_list payments_job”)工作:試一試:p =多處理。過程(目標= GetData, arg遊戲= (job_Context、“0030161”)) p.start () processes.append除了異常(p)為例:狀態=消息= f“CL_Failed工作{工作}失敗由於一個例外。有關更多信息,請參閱磚。錯誤:repr(特異){}”job_context.logger_error(消息)提高p的交貨流程:p.join ()

錯誤日誌中我可以看到如下

文件“命令- 2328844487947527 > <”,13號線,在GetData打印(payments_df)文件“/磚/ / python / pyspark / sql / dataframe火花。py”, 452行,在__repr__回報”DataFrame [% s]“% (“,”。加入(“% s: % s“% c c self.dtypes))文件“/磚/ / python / pyspark / sql / dataframe火花。py”, 1040行,在dtypes返回(str (f.name) f.dataType.simpleString())在self.schema f。字段)文件“/磚/ / python / pyspark / sql / dataframe火花。py”, 257行,在模式的自我。_schema = _parse_datatype_json_string (self._jdf.schema () . json())文件“/磚/火花/ python / lib / py4j-0.10.9-src.zip / py4j / java_gateway。py”, 1305行,在__call__回答,自我。gateway_client,自我。target_id self.name)文件“/磚/火花/ python / pyspark / sql /跑龍套。py”, 127行,在裝飾返回f(*, * *千瓦)文件”/磚/火花/ python / lib / py4j-0.10.9-src.zip / py4j /協議。py”, 332行,在get_return_value格式(target_id”。”,名字,py4j.protocol值))。Py4JError:調用o1361.schema時發生一個錯誤。跟蹤:py4j。Py4JException:方法模式([])並不存在

這是完全罰款如果不是運行在一個單獨的進程

任何想法或幫助將不勝感激

[Hubert_Dudek1]

火花,默認情況下,在一個分布式的方式工作,這個過程並行使用本機火花。

它將使用所有的核心;每個核心將處理一個分區和寫數據庫。

selec也一樣。

舉個例子,您可以使用下麵的代碼來讀取SQL並行:

表=(火花。讀.format (jdbc) .option (“url”,“< jdbc_url >”) .option(“數據表”,“< table_name >”) .option(“用戶”,“<用戶名>”).option(“密碼”,“<密碼>”)#一列,可以使用均勻分布範圍的值,可用於並行.option (“partitionColumn”、“< partition_key >”) #最小值將數據與partitionColumn .option(“下界”、“< min_value >”) #馬克斯值將數據與partitionColumn .option (“upperBound”、“< max_value >”) #分區數量將數據分發到。使用sc.defaultParallelism設定數量的核.option (“numPartitions”, sc.defaultParallelism) .load ()

[vanepet]

如何實現這一目標我想運行在並行查詢不同的表嗎?

就像

DF =從表,集團選擇* = 1

DF2 = Select *從表b組= 1

[Soma]

@Peter Vanelli

檢查是否這有助於

https://medium.com/analytics-vidhya/horizontal-parallelism-with-pyspark-d05390aa1df5

[huyd]

這可能幫助https://dustinvannoy.com/2022/05/06/parallel-ingest-spark-notebook/

編寫一個函數來通過你的表列表通過數組,為線程隊列