0737-1.6.1-CDSW分散式計算

  • 2020 年 1 月 14 日
  • 筆記

作者：冉南陽

簡介

Cloudera Data Science Workbench為從單個互動式會話中啟動多個稱為workers的引擎實例提供了基本支援。任何R或Python會話均可用於生成workers。這些工作程式可以配置為在啟動時運行腳本（例如Python文件）或命令中。

可以使用 launch_workers函數。以及另外兩個函數是list_workers 和 list_workers。來自所有工作程式的輸出將顯示在啟動它們的會話的控制台中。會話退出時，這些workers將終止。

Python示例

Python分散式運算基本語法

import cdsw  workers=cdsw.launch_workers(n=2,cpu=0.2,memory=0.5,code="print('Hello from a CDSW Woker')")
  • n（int） -要啟動的引擎數。
  • cpu（float） -要分配給引擎的CPU內核數。
  • 記憶體（float） -分配給引擎的記憶體的千兆位元組數。
  • nvidia_gpu（int，optional） -要分配給引擎的GPU數量。
  • kernel（str，optional） -內核。可以是r、 python2、 python3或scala。
  • script（str，optional） -工作者啟動後應立即執行的Python源文件的名稱。
  • code（str，optional） -引擎在啟動後應立即執行的Python程式碼。如果指定了腳本，則程式碼將被忽略。
  • env（dict，optional） -在引擎中設置的環境變數。

list_workers

cdsw.list_workers()

stop_worker

cdsw.stop_workers()

編輯Python文件master.py；在Master引擎中，該腳本將啟動三個工作程式並接受來自Workers的傳入連接。

# master.py  import cdsw, socket  # Launch two CDSW workers. These are engines that will run in  # the same project, execute a given code or script, and exit.  workers = cdsw.launch_workers(n=3, cpu=0.2, memory=0.5, script="worker.py")  # Listen on TCP port 6000  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  s.bind(("0.0.0.0", 6000))  s.listen(1)  # Accept two connections, one from each worker. Workers will  # execute worker.py.  conn, addr = s.accept()  for i in range(3):      # Receive a message from each worker and return a response.      data = conn.recv(20)      if not data: break      print("Master received:", data)      conn.send("Hello From Server!".encode())  conn.close()

編輯worker.py文件；Workers將執行腳本中的命令，並返回給Master。

# worker.py  import os, socket  # Open a TCP connection to the master.  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  s.connect((os.environ["CDSW_MASTER_IP"], 6000))  # Send some data and receive a response.  s.send("Hello From Worker!".encode())  data = s.recv(1024)  s.close()  print("Worker received:", data)

執行master.py，可以看到執行了三次worker.py。

R語言示例

基本語法

library("cdsw")  workers <- launch.workers(n=2,cpu=0.2,memory=0.5, env="",code="print('Hello From a CDSW Workers')")

由於解析 launch.workers函數的bug,因此定義env 參數。如果未定義env 參數，env參數在內部序列化為與Cloudera Data Science Workbench不兼容的格式。此bug不會影響Python引擎。

編寫master.r文件，在Master引擎中，該腳本將啟動兩個工作程式並接受來自Workers的傳入連接。

# master.r  library("cdsw")  # Launch two CDSW workers. These are engines that will run in  # the same project, execute a given code or script, and exit.  workers <- launch.workers(n=2, cpu=0.2, memory=0.5, env="", script="worker.r")  # Accept two connections, one from each worker. Workers will  # execute worker.r.  for(i in c(1,2)) {    # Receive a message from each worker and return a response.    con <- socketConnection(host="0.0.0.0", port = 6000, blocking=TRUE, server=TRUE, open="r+")    data <- readLines(con, 1)    print(paste("Server received:", data))    writeLines("Hello from master!", con)    close(con)  }

編寫worker.r文件；Workers將執行以下命令，並返回到Master。

# worker.r  print(Sys.getenv("CDSW_MASTER_IP"))  con <- socketConnection(host=Sys.getenv("CDSW_MASTER_IP"), port = 6000, blocking=TRUE, server=FALSE, open="r+")  write_resp <- writeLines("Hello from Worker", con)  server_resp <- readLines(con, 1)  print(paste("Worker received:  ", server_resp))  close(con)

執行master.r，可以看到執行了兩次worker.r

參考文檔

https://docs.cloudera.com/documentation/data-science-workbench/1-6-x/topics/cdsw_parallel_computing.html

Tags:

VirMach 便宜 VPS