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
