pyspark spark onExecutorMetricsUpdate无法获取executor_id

dffbzjpn  于 2024-01-06  发布在  Spark
关注(0)|答案(2)|浏览(159)

在下面的代码中，在pyspark 2.x中，使用executorMetricsUpdate.execId()总是给出一个数字形式的执行器ID，例如'31'（虽然是字符串类型）。但是从pyspark 2.x升级到3.4.1之后，executor_id总是'driver'。为什么？我依赖于这个executor_id来创建一个关于内存使用情况的统计报告。有没有一种方法可以访问该指标所针对的executor？

  1. def onExecutorMetricsUpdate(self, executorMetricsUpdate):
  2.     executor_id = executorMetricsUpdate.execId()

字符串
更新：
在我的配置中，我有：

  1. "spark.master": "yarn",

型
此外，如果我修改我的spark作业以返回它正在运行的机器，我可以看到集群中的多台机器已经被使用。
如果我点击Spark API端点

  1. ...stages/<stage_id>/<attempt_id>/taskList

型
我可以在不同的主机上看到多个任务，但所有的内存指标（例如peakExecutionMemory）都设置为0：

  1. "taskId" : 0,
  2.   "index" : 0,
  3.   "attempt" : 0,
  4.   "partitionId" : 0,
  5.   "launchTime" : "2023-11-30T13:48:28.018GMT",
  6.   "duration" : 11850,
  7.   "executorId" : "127",
  8.   "host" : "xxxxxxx",
  9.   "status" : "SUCCESS",
  10.   "taskLocality" : "PROCESS_LOCAL",
  11.   "speculative" : false,
  12.   "accumulatorUpdates" : [ ],
  13.   "taskMetrics" : {
  14.     "executorDeserializeTime" : 750,
  15.     "executorDeserializeCpuTime" : 666813421,
  16.     "executorRunTime" : 10990,
  17.     "executorCpuTime" : 129812934,
  18.     "resultSize" : 1366,
  19.     "jvmGcTime" : 0,
  20.     "resultSerializationTime" : 4,
  21.     "memoryBytesSpilled" : 0,
  22.     "diskBytesSpilled" : 0,
  23.     "peakExecutionMemory" : 0,
  24.     "inputMetrics" : {
  25.       "bytesRead" : 0,
  26.       "recordsRead" : 0
  27.     },

型
如果我点击/executors端点，我只看到“驱动程序”
更新2：
我在一台用户机器上使用pyspark运行了我的spark应用程序（与spark worker机器不同）：

  1. # this utility function which sets a few conf_dict variables
  2. def custom_create_context(..):
  3.   conf = pyspark.SparkConf().setAll(conf_dict.items())
  4.   ss = pyspark.sql.SparkSession.builder.config(conf=conf).getOrCreate()
  5.   sc = ss.sparkContext
  6.   return sc, ss
  8.   # called by:
  10. sc, ss = custom_create_context(conf_dict={
  11.     "spark.app.name":                       "ftest1",
  12.     "spark.driver.memory":                  "2g",
  13.     "spark.executor.memory":                "2g",
  14.     "spark.executor.cores":                 "1", 
  15.     "spark.task.cpus":                      "1", 
  16.     "spark.dynamicAllocation.maxExecutors": "1000",
  17.     "spark.master": "yarn",
  18. }, show_report=True)
  20. # register listener
  22. listener = MemoryReporter(sc, report_dir)
  23. sc._jsc.sc().addSparkListener(listener)
  25. # submit work to spark
  26. x = [some data]
  27. result = sc.parallelize(x, numSlices=len(x)).map(lambda x: myfunc(x)).collect()
  28. print(result)

型
conf_dict变量：

  1. spark.app.name: xxx
  2. spark.driver.memory: 2g
  3. spark.executor.memory: 2g
  4. spark.executor.cores: 1
  5. spark.task.cpus: 1
  6. spark.dynamicAllocation.maxExecutors: 1000
  7. spark.yarn.appMasterEnv.SPARK_WITH_NFS: true
  8. spark.executorEnv.SPARK_WITH_NFS: true
  9. spark.driverEnv.SPARK_WITH_NFS: true
  10. spark.master: yarn
  11. spark.submit.deployMode: client
  12. spark.dynamicAllocation.enabled: true
  13. spark.shuffle.service.enabled: true
  14. spark.dynamicAllocation.executorIdleTimeout: 120s
  15. spark.dynamicAllocation.cachedExecutorIdleTimeout: 300s
  16. spark.driver.cores: 4
  17. spark.driver.maxResultSize: 32g
  18. spark.yarn.am.waitTime: 30s
  19. spark.ui.showConsoleProgress: true
  20. spark.worker.ui.retainedExecutors: 5000
  21. spark.ui.retainedDeadExecutors: 5000
  22. spark.eventLog.enabled: true
  23. spark.eventLog.dir: hdfs:///spark-history
  24. spark.starvation.timeout: 90s
  25. spark.local.dir: xxx
  26. spark.python.worker.memory: 1024m
  27. spark.pyspark.python: xxx
  28. spark.yarn.appMasterEnv.PYSPARK_PYTHON: xxx
  29. spark.executorEnv.PYSPARK_PYTHON: xx
  30. spark.driverEnv.PYSPARK_PYTHON: xxx
  31. spark.yarn.appMasterEnv.VAULT_BUE_CACHE: xxx
  32. spark.executorEnv.VAULT_BUE_CACHE: xxx
  33. spark.driverEnv.VAULT_BUE_CACHE: xxx
  35. spark.redaction.regex: xxx
  36. spark.yarn.appMasterEnv.SHELL: /bin/bash
  37. spark.executorEnv.SHELL: /bin/bash
  38. spark.driverEnv.SHELL: /bin/bash
  40. spark.yarn.appMasterEnv.RCFILE: xxx
  41. spark.executorEnv.RCFILE: xxx
  42. spark.driverEnv.RCFILE: xxx
  44. spark.yarn.appMasterEnv.TERM_PROGRAM_VERSION: 3.2a
  45. spark.executorEnv.TERM_PROGRAM_VERSION: 3.2a
  46. spark.driverEnv.TERM_PROGRAM_VERSION: 3.2a
  49. spark.yarn.appMasterEnv.BSTINPUTS: .:/data/app/latex/bst/:/usr/share/texmf/bibtex/bst//:
  50. spark.executorEnv.BSTINPUTS: .:/data/app/latex/bst/:/usr/share/texmf/bibtex/bst//:
  51. spark.driverEnv.BSTINPUTS: .:/data/app/latex/bst/:/usr/share/texmf/bibtex/bst//:
  53. spark.yarn.appMasterEnv.LOGNAME: xxx
  54. spark.executorEnv.LOGNAME: xxx
  55. spark.driverEnv.LOGNAME: xxx
  56. spark.yarn.appMasterEnv.XDG_SESSION_TYPE: tty
  57. spark.executorEnv.XDG_SESSION_TYPE: tty
  58. spark.driverEnv.XDG_SESSION_TYPE: tty
  59. spark.yarn.appMasterEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  60. spark.executorEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  61. spark.driverEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  63. spark.yarn.appMasterEnv.VIRTUAL_ENV: xxx
  64. spark.executorEnv.VIRTUAL_ENV: xxx
  65. spark.driverEnv.VIRTUAL_ENV: xxx
  66. spark.yarn.appMasterEnv.VAULT_TMPFS: xxx
  67. spark.executorEnv.VAULT_TMPFS: xxx
  68. spark.driverEnv.VAULT_TMPFS: xxx
  69. spark.yarn.appMasterEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  70. spark.executorEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  71. spark.driverEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  72. spark.yarn.appMasterEnv.ORACLE_HOME: /app/oracle
  73. spark.executorEnv.ORACLE_HOME: /app/oracle
  74. spark.driverEnv.ORACLE_HOME: /app/oracle
  76. spark.yarn.appMasterEnv.SSH_CONNECTION: xxx
  77. spark.executorEnv.SSH_CONNECTION: xxx
  78. spark.driverEnv.SSH_CONNECTION: xxx
  79. spark.yarn.appMasterEnv.HOMEBREW
  80. spark.driverEnv.GIT_BASE: xxx
  81. spark.yarn.appMasterEnv.TERMINFO: /usr/share/terminfo
  82. spark.executorEnv.TERMINFO: /usr/share/terminfo
  83. spark.driverEnv.TERMINFO: /usr/share/terminfo
  84. spark.yarn.appMasterEnv.TERM: screen
  85. spark.executorEnv.TERM: screen
  86. spark.driverEnv.TERM: screen
  87. spark.yarn.appMasterEnv.USER: xxx
  88. spark.executorEnv.USER: xxx
  89. spark.driverEnv.USER: xxx
  91. spark.yarn.appMasterEnv.LC_TERMINAL_VERSION: 3.4.20
  92. spark.executorEnv.LC_TERMINAL_VERSION: 3.4.20
  93. spark.driverEnv.LC_TERMINAL_VERSION: 3.4.20
  94. spark.yarn.appMasterEnv.CONSUL_HTTP_SSL: true
  95. spark.executorEnv.CONSUL_HTTP_SSL: true
  96. spark.driverEnv.CONSUL_HTTP_SSL: true
  97. spark.yarn.appMasterEnv.SHLVL: 2
  98. spark.executorEnv.SHLVL: 2
  99. spark.driverEnv.SHLVL: 2
  101. spark.yarn.appMasterEnv.XDG_SESSION_ID: 3
  102. spark.executorEnv.XDG_SESSION_ID: 3
  103. spark.driverEnv.XDG_SESSION_ID: 3
  104. spark.yarn.appMasterEnv.PROFILE_DEBUG: no
  105. spark.executorEnv.PROFILE_DEBUG: no
  106. spark.driverEnv.PROFILE_DEBUG: no
  107. spark.yarn.appMasterEnv.XDG_RUNTIME_DIR: xxx
  108. spark.executorEnv.XDG_RUNTIME_DIR: xx
  109. spark.driverEnv.XDG_RUNTIME_DIR: xxx
  110. spark.yarn.appMasterEnv.PROCPS_USERLEN: 16
  111. spark.executorEnv.PROCPS_USERLEN: 16
  112. spark.driverEnv.PROCPS_USERLEN: 16
  113. spark.yarn.appMasterEnv.SSH_CLIENT: xxx
  114. spark.executorEnv.SSH_CLIENT: xxx
  115. spark.driverEnv.SSH_CLIENT: xxx
  116. spark.yarn.appMasterEnv.REQUESTS_CA_BUNDLE: xx
  117. spark.executorEnv.REQUESTS_CA_BUNDLE: xxx
  118. spark.driverEnv.REQUESTS_CA_BUNDLE: xxx
  120. spark.yarn.appMasterEnv.DOCKER_HOST: unix:///xx
  121. spark.executorEnv.DOCKER_HOST: unix:///xxx
  122. spark.driverEnv.DOCKER_HOST: unix:///xxx
  123. spark.yarn.appMasterEnv.TNS_ADMIN: xxx
  124. spark.executorEnv.TNS_ADMIN: xxx
  125. spark.driverEnv.TNS_ADMIN: xxx
  127. spark.yarn.appMasterEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  128. spark.executorEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  129. spark.driverEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  130. spark.yarn.appMasterEnv.PATH: xxx
  131. spark.executorEnv.PATH: xxx
  132. spark.driverEnv.PATH: xxx
  134. spark.yarn.appMasterEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=/xxx
  135. spark.executorEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=/xxx
  136. spark.driverEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=/xxx
  138. spark.yarn.appMasterEnv.SSH_TTY: /dev/pts/0
  139. spark.executorEnv.SSH_TTY: /dev/pts/0
  140. spark.driverEnv.SSH_TTY: /dev/pts/0
  141. spark.yarn.appMasterEnv.OLDPWD: /xx
  142. spark.executorEnv.OLDPWD: x
  143. spark.driverEnv.OLDPWD: xx
  145. spark.yarn.appMasterEnv.CONSUL_HTTP_ADDR: xxx
  146. spark.executorEnv.CONSUL_HTTP_ADDR: xxx
  147. spark.driverEnv.CONSUL_HTTP_ADDR: xxx
  149. spark.yarn.appMasterEnv.SSL_CERT_FILE: xxx
  150. spark.executorEnv.SSL_CERT_FILE: xxx
  151. spark.driverEnv.SSL_CERT_FILE: xx
  153. spark.yarn.appMasterEnv.CONSUL_TOKEN_JSON: xxx
  154. spark.executorEnv.CONSUL_TOKEN_JSON: xxx
  155. spark.driverEnv.CONSUL_TOKEN_JSON: xx
  156. spark.yarn.appMasterEnv.LOGNAME: x
  157. spark.executorEnv.LOGNAME: x
  158. spark.driverEnv.LOGNAME: x
  159. spark.yarn.appMasterEnv.XDG_SESSION_TYPE: tty
  160. spark.executorEnv.XDG_SESSION_TYPE: tty
  161. spark.driverEnv.XDG_SESSION_TYPE: tty
  162. spark.yarn.appMasterEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  163. spark.executorEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  164. spark.driverEnv.TEXINPUTS: .:/data/app/latex/style/:/usr/share/texmf/tex//:
  165. spark.yarn.appMasterEnv.CONSUL_LEASE_FILE: xxx
  166. spark.executorEnv.CONSUL_LEASE_FILE: xxx
  167. spark.driverEnv.CONSUL_LEASE_FILE: xxx
  168. spark.yarn.appMasterEnv.CONSUL_HTTP_TOKEN_ACCESSOR: xxx
  169. spark.executorEnv.CONSUL_HTTP_TOKEN_ACCESSOR: xxx
  170. spark.driverEnv.CONSUL_HTTP_TOKEN_ACCESSOR: xxx
  172. spark.yarn.appMasterEnv.CONSUL_HTTP_TOKEN: xxx
  173. spark.executorEnv.CONSUL_HTTP_TOKEN: xxx
  174. spark.driverEnv.CONSUL_HTTP_TOKEN: xxx
  176. spark.yarn.appMasterEnv.VIRTUAL_ENV: xxx
  177. spark.executorEnv.VIRTUAL_ENV: xxx
  178. spark.driverEnv.VIRTUAL_ENV: xxx
  179. spark.yarn.appMasterEnv.VAULT_TMPFS: xxx
  180. spark.executorEnv.VAULT_TMPFS: xxx
  181. spark.driverEnv.VAULT_TMPFS: xxx
  182. spark.yarn.appMasterEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  183. spark.executorEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  184. spark.driverEnv.BIBINPUTS: .:/data/app/latex/:/usr/share/texmf/bibtex/bib//:
  185. spark.yarn.appMasterEnv.ORACLE_HOME: /app/oracle
  186. spark.executorEnv.ORACLE_HOME: /app/oracle
  187. spark.driverEnv.ORACLE_HOME: /app/oracle
  189. spark.yarn.appMasterEnv.SSH_CONNECTION: xxx
  190. spark.executorEnv.SSH_CONNECTION: xxx
  191. spark.driverEnv.SSH_CONNECTION: xxx
  192. spark.yarn.appMasterEnv.HOMEBREW_NO_AUTO_UPDATE: 1
  193. spark.executorEnv.HOMEBREW_NO_AUTO_UPDATE: 1
  194. spark.driverEnv.HOMEBREW_NO_AUTO_UPDATE: 1
  195. spark.yarn.appMasterEnv.CONSUL_HTTP_TOKEN_FILE: xxx
  196. spark.executorEnv.CONSUL_HTTP_TOKEN_FILE: xxx
  197. spark.driverEnv.CONSUL_HTTP_TOKEN_FILE: xxx
  199. spark.executorEnv.CONSUL_HTTP_TOKEN_FILE: xxx                                                                                                                             
  200. spark.driverEnv.CONSUL_HTTP_TOKEN_FILE: xxx
  201. spark.yarn.appMasterEnv.XDG_SESSION_CLASS: user
  202. spark.executorEnv.XDG_SESSION_CLASS: user
  203. spark.driverEnv.XDG_SESSION_CLASS: user
  204. spark.yarn.appMasterEnv.GIT_BASE: xxx
  205. spark.executorEnv.GIT_BASE: xxx
  206. spark.driverEnv.GIT_BASE: xxx
  207. spark.yarn.appMasterEnv.TERMINFO: /usr/share/terminfo
  208. spark.executorEnv.TERMINFO: /usr/share/terminfo
  209. spark.driverEnv.TERMINFO: /usr/share/terminfo
  210. spark.yarn.appMasterEnv.TERM: screen
  211. spark.executorEnv.TERM: screen
  212. spark.driverEnv.TERM: screen
  213. spark.yarn.appMasterEnv.USER: xxx
  214. spark.executorEnv.USER: xxx
  215. spark.driverEnv.USER: xxx
  217. spark.yarn.appMasterEnv.CONSUL_HTTP_SSL: true
  218. spark.executorEnv.CONSUL_HTTP_SSL: true
  219. spark.driverEnv.CONSUL_HTTP_SSL: true
  220. spark.yarn.appMasterEnv.SHLVL: 2
  221. spark.executorEnv.SHLVL: 2
  222. spark.driverEnv.SHLVL: 2
  224. spark.yarn.appMasterEnv.GIT_EXAMPLE: xxx
  225. spark.executorEnv.GIT_EXAMPLE: xxx
  226. spark.driverEnv.GIT_EXAMPLE: xxx
  227. spark.yarn.appMasterEnv.XDG_SESSION_ID: 3
  228. spark.executorEnv.XDG_SESSION_ID: 3
  229. spark.driverEnv.XDG_SESSION_ID: 3
  230. spark.yarn.appMasterEnv.PROFILE_DEBUG: no
  231. spark.executorEnv.PROFILE_DEBUG: no
  232. spark.driverEnv.PROFILE_DEBUG: no
  233. spark.yarn.appMasterEnv.XDG_RUNTIME_DIR: xxx
  234. spark.executorEnv.XDG_RUNTIME_DIR: xxx
  235. spark.driverEnv.XDG_RUNTIME_DIR: xxx
  236. spark.yarn.appMasterEnv.PROCPS_USERLEN: 16
  237. spark.executorEnv.PROCPS_USERLEN: 16
  238. spark.driverEnv.PROCPS_USERLEN: 16
  239. spark.yarn.appMasterEnv.PYTHONBREAKPOINT: pudb.set_trace
  240. spark.executorEnv.PYTHONBREAKPOINT: pudb.set_trace
  241. spark.driverEnv.PYTHONBREAKPOINT: pudb.set_trace
  242. spark.yarn.appMasterEnv.SSH_CLIENT: xxx
  243. spark.executorEnv.SSH_CLIENT: xx
  244. spark.driverEnv.SSH_CLIENT: xxx
  245. spark.yarn.appMasterEnv.REQUESTS_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  246. spark.executorEnv.REQUESTS_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  247. spark.driverEnv.REQUESTS_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  249. spark.yarn.appMasterEnv.DOCKER_HOST: unix:///xxx
  250. spark.executorEnv.DOCKER_HOST: unix:///xxx
  251. spark.driverEnv.DOCKER_HOST: unix:///xx
  252. spark.yarn.appMasterEnv.TNS_ADMIN: xxx
  253. spark.executorEnv.TNS_ADMIN: xxx
  254. spark.driverEnv.TNS_ADMIN: xxx
  256. spark.yarn.appMasterEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  257. spark.executorEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  258. spark.driverEnv.XDG_DATA_DIRS: /usr/share/gnome:/usr/local/share/:/usr/share/
  259. spark.yarn.appMasterEnv.PATH: xxx
  260. spark.executorEnv.PATH: xxx
  261. spark.driverEnv.PATH: xxx
  263. spark.yarn.appMasterEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=xxx
  264. spark.executorEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=xx
  265. spark.driverEnv.DBUS_SESSION_BUS_ADDRESS: unix:path=xxx
  267. spark.yarn.appMasterEnv.SSH_TTY: /dev/pts/0
  268. spark.executorEnv.SSH_TTY: /dev/pts/0
  269. spark.driverEnv.SSH_TTY: /dev/pts/0
  270. spark.yarn.appMasterEnv.OLDPWD: xxx
  271. spark.executorEnv.OLDPWD: xxx
  272. spark.driverEnv.OLDPWD: xxx
  274. spark.yarn.appMasterEnv.CONSUL_HTTP_ADDR: xxx
  275. spark.executorEnv.CONSUL_HTTP_ADDR: xx
  276. spark.driverEnv.CONSUL_HTTP_ADDR: xxx
  278. spark.yarn.appMasterEnv.SSL_CERT_FILE: /etc/ssl/certs/ca-certificates.crt
  279. spark.executorEnv.SSL_CERT_FILE: /etc/ssl/certs/ca-certificates.crt
  280. spark.driverEnv.SSL_CERT_FILE: /etc/ssl/certs/ca-certificates.crt
  281. spark.yarn.appMasterEnv.CURL_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  282. spark.executorEnv.CURL_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  283. spark.driverEnv.CURL_CA_BUNDLE: /etc/ssl/certs/ca-certificates.crt
  286. spark.yarn.appMasterEnv.PYSPARK_RDD_CODE_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  287. spark.executorEnv.PYSPARK_RDD_CODE_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  288. spark.driverEnv.PYSPARK_RDD_CODE_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  289. spark.yarn.appMasterEnv.PYSPARK_RDD_DATA_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  290. spark.executorEnv.PYSPARK_RDD_DATA_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  291. spark.driverEnv.PYSPARK_RDD_DATA_SERIALIZER: pyspark.serializers.CloudPickleSerializer
  292. spark.yarn.appMasterEnv.HADOOP_HOME: /.../hadoop/current
  293. spark.executorEnv.HADOOP_HOME: xxx
  294. spark.driverEnv.HADOOP_HOME: xxx
  295. spark.yarn.appMasterEnv.HADOOP_COMMON_HOME: xxx
  296. spark.executorEnv.HADOOP_COMMON_HOME: xxx
  297. spark.driverEnv.HADOOP_COMMON_HOME: xxx
  298. spark.yarn.appMasterEnv.HADOOP_HDFS_HOME: xxx
  299. spark.executorEnv.HADOOP_HDFS_HOME: xxx
  300. spark.driverEnv.HADOOP_HDFS_HOME: xxx
  301. spark.yarn.appMasterEnv.HADOOP_YARN_HOME: xxx
  302. spark.executorEnv.HADOOP_YARN_HOME: xxx
  303. spark.driverEnv.HADOOP_YARN_HOME: xxx
  304. spark.yarn.appMasterEnv.HADOOP_MAPRED_HOME: xxx
  305. spark.executorEnv.HADOOP_MAPRED_HOME: xxx
  306. spark.driverEnv.HADOOP_MAPRED_HOME: xxx
  307. spark.yarn.appMasterEnv.HADOOP_CONF_DIR: xxx
  308. spark.executorEnv.HADOOP_CONF_DIR: xxx
  309. spark.driverEnv.HADOOP_CONF_DIR: xxx
  310. spark.yarn.appMasterEnv.OMP_NUM_THREADS: 1
  311. spark.executorEnv.OMP_NUM_THREADS: 1
  312. spark.driverEnv.OMP_NUM_THREADS: 1
  313. spark.yarn.appMasterEnv.SPARK_HOME: xxx
  314. spark.executorEnv.SPARK_HOME: xxx
  315. spark.driverEnv.SPARK_HOME: xxx

型
更新3：
MemoryReporter是一个自定义类，它接收https://spark.apache.org/docs/2.3.3/api/scala/index.html#org.apache.spark.scheduler.SparkListener中描述的事件：

  1. API_URL_VAR = "spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY_URI_BASES"
  3. pyspark_v3 = pyspark.__version__.startswith("3.")
  5. class SparkListener(object):
  6.     """
  7.     https://spark.apache.org/docs/2.3.3/api/scala/index.html#org.apache.spark.scheduler.SparkListener
  8.     """
  10.     def onApplicationEnd(self, applicationEnd):
  11.         pass
  13.     def onApplicationStart(self, applicationStart):
  14.         pass
  16.     def onBlockManagerAdded(self, blockManagerAdded):
  17.         pass
  19.     def onBlockManagerRemoved(self, blockManagerRemoved):
  20.         pass
  22.     def onBlockUpdated(self, blockUpdated):
  23.         pass
  25.     def onEnvironmentUpdate(self, environmentUpdate):
  26.         pass
  28.     def onExecutorAdded(self, executorAdded):
  29.         pass
  31.     def onExecutorBlacklisted(self, executorBlacklisted):
  32.         pass
  34.     def onExecutorMetricsUpdate(self, executorMetricsUpdate):
  35.         pass
  37.     def onExecutorRemoved(self, executorRemoved):
  38.         pass
  40.     def onExecutorUnblacklisted(self, executorUnblacklisted):
  41.         pass
  43.     def onJobEnd(self, jobEnd):
  44.         pass
  46.     def onJobStart(self, jobStart):
  47.         pass
  49.     def onNodeBlacklisted(self, nodeBlacklisted):
  50.         pass
  52.     def onNodeUnblacklisted(self, nodeUnblacklisted):
  53.         pass
  55.     def onOtherEvent(self, event):
  56.         pass
  58.     def onSpeculativeTaskSubmitted(self, speculativeTask):
  59.         pass
  61.     def onStageCompleted(self, stageCompleted):
  62.         pass
  64.     def onStageSubmitted(self, stageSubmitted):
  65.         pass
  67.     def onTaskEnd(self, taskEnd):
  68.         pass
  70.     def onTaskGettingResult(self, taskGettingResult):
  71.         pass
  73.     def onTaskStart(self, taskStart):
  74.         pass
  76.     def onUnpersistRDD(self, unpersistRDD):
  77.         pass
  79.     class Java:
  80.         implements = ["org.apache.spark.scheduler.SparkListenerInterface"]
  82. def _scala_iterable_to_py_list(scala_iterable):
  83.     if isinstance(scala_iterable, Iterable):
  84.         return list(scala_iterable)
  86.     py_list = []
  87.     it = scala_iterable.toIterator()
  88.     while it.hasNext():
  89.         py_list.append(it.next())
  90.     return py_list
  92. def _scala_set_to_py_set(scala_set):
  93.     return set(_scala_iterable_to_py_list(scala_set))
  95. class MemoryReporter(SparkListener):
  96.     def __init__(self, sc, report_dir):
  97.         self.logger = logging.getLogger(self.__class__.__name__)
  99.         self._sc = sc
  100.         self._report_dir = os.environ.get(SPARK_MEMORY_REPORT_DIR_ENVIRON, report_dir)
  102.         self._spark_conf = sc.getConf()
  104.         self._username = self._sc._jvm.java.lang.System.getProperty("user.name")
  105.         self._application_id = self._spark_conf.get("spark.app.id", None)
  107.         spark_executor_memory = self._spark_conf.get("spark.executor.memory", None)
  108.         self._requested_limit_byte = self._parse_memory_from_conf(spark_executor_memory)
  109.         self._actual_limit_byte = 0  # will be updated once job is running
  111.         self._executor_task = {}  # type: Dict[int, int]
  112.         self._task_memory = {}  # type: Dict[int, int]
  113.         self._stage_tasks = {}  # type: Dict[int, Set[int]]
  114.         self._stage_last_attempt_id = {}
  115.         self._active_jobs = 0
  117.     def _parse_memory_from_conf(self, memory):
  118.         pass # edited out for briefness
  120.     def _reset_memory_stats(self):
  121.         pass # edited out for briefness
  123.     def _format_memory_usage(self, x):
  124.         pass # edited out for briefness
  126.     def _print_report(self, task_memory):
  127.         pass # edited out for briefness
  129.     def update_stage_info_v2(self):
  130.         status_store = self._sc._jsc.sc().statusStore()
  132.         all_jobs = _scala_iterable_to_py_list(status_store.jobsList(None))
  133.         all_stage_ids = []
  134.         for job in all_jobs:
  135.             all_stage_ids += _scala_iterable_to_py_list(job.stageIds())
  137.         for stage_id in all_stage_ids:
  138.             if stage_id in self._stage_tasks:
  139.                 continue
  141.             stage_data = _scala_iterable_to_py_list(status_store.stageData(stage_id, True))
  142.             if not stage_data:
  143.                 continue
  145.             last_stage_data = stage_data[-1]
  146.             stage_status = last_stage_data.status().toString()
  147.             if stage_status != "ACTIVE" and stage_status != "PENDING" and last_stage_data.tasks().isDefined():
  148.                 stage_tasks = stage_data[-1].tasks().get()
  149.                 stage_task_ids = _scala_set_to_py_set(stage_tasks.keys())
  150.                 self._stage_tasks[stage_id] = stage_task_ids
  152.     def update_stage_info_v3(self):
  153.         """
  154.         In pyspark 3 we don't have:
  155.         - stage_info.status
  156.         - status_store.stageData(stage_id, True)
  157.         - a way to get task info from the stage (only via REST API)
  158.         """
  159.         status_store = self._sc._jsc.sc().statusStore()
  160.         all_jobs = _scala_iterable_to_py_list(status_store.jobsList(None))
  161.         all_stage_ids = []
  162.         for job in all_jobs:
  163.             all_stage_ids += _scala_iterable_to_py_list(job.stageIds())
  165.         for stage_id in all_stage_ids:
  166.             if stage_id in self._stage_tasks:
  167.                 continue
  169.             app_id = self._sc.applicationId
  170.             proxy_host = self._sc.getConf().get(API_URL_VAR)
  171.             if not proxy_host:
  172.                 continue
  174.             stages_api_url = f"{proxy_host}/api/v1/applications/{app_id}/stages/{stage_id}"
  176.             try:
  177.                 response = requests.get(stages_api_url)
  178.                 stages_data = response.json()
  179.             except Exception as ex:
  180.                 self.logger.warning(f"Failed getting stages_data from API Url: {stages_api_url} ex: {ex}")
  181.                 continue
  183.             last_stage_attempt = stages_data[-1]
  184.             last_stage_status = last_stage_attempt.get("status")
  185.             if last_stage_status and last_stage_status != "ACTIVE" and last_stage_status != "PENDING":
  186.                 attempt_id = last_stage_attempt.get("attemptId")
  187.                 self._stage_last_attempt_id[stage_id] = attempt_id
  188.                 if attempt_id is not None:  # careful as it can be 0 (evaluates to False)
  189.                     tasks_info_api_url = (
  190.                         f"{proxy_host}/api/v1/applications/{app_id}/stages/{stage_id}/{attempt_id}/taskList"
  191.                     )
  192.                     try:
  193.                         response2 = requests.get(tasks_info_api_url)
  194.                         tasks_info_data = response2.json()
  195.                         self.logger.debug(f"tasks_info_api_url: {tasks_info_api_url}")
  196.                         task_ids = [t["taskId"] for t in tasks_info_data]
  197.                         self._stage_tasks[stage_id] = task_ids
  199.                     except Exception as ex:
  200.                         self.logger.warning(f"Failed getting tasks_info_data API url: {tasks_info_api_url} ex: {ex}")
  201.                         continue
  203.     def _update_stage_tasks(self):
  204.         if pyspark_v3:
  205.             self.update_stage_info_v3()
  206.         else:
  207.             self.update_stage_info_v2()
  209.     def _save_csv(self, job_id, df):
  210.         pass # edited out for briefness
  212.     def _save_report(self, job_id, task_memory):
  213.         pass # edited out for briefness
  215.     def onTaskStart(self, taskStart):
  216.         task_id = taskStart.taskInfo().taskId()
  217.         executor_id = taskStart.taskInfo().executorId()
  219.         self.logger.debug("setting executor '{}' to task '{}'".format(executor_id, task_id))
  220.         self._executor_task[executor_id] = task_id
  222.     def onExecutorMetricsUpdate(self, executorMetricsUpdate):
  223.         executor_id = executorMetricsUpdate.execId()  # returns 'driver' in pyspark > 3 but a number in pyspark < 2
  224.         if pyspark_v3:
  225.             # https://spark.apache.org/docs/3.1.1/monitoring.html
  226.             # https://github.com/apache/spark/blob/v3.1.1/core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala#L196
  227.             memory_byte = 0
  228.             metrics = executorMetricsUpdate.executorUpdates().values().toList()  # Scala view to values
  229.             metrics = [metrics.apply(i) for i in range(metrics.size())]
  230.             for m in metrics:
  231.                 memory_byte = max(memory_byte, m.getMetricValue("JVMHeapMemory"))
  232.         else:
  233.             # Backward compatible. e.g. v2.3.4
  234.             memory_byte = executorMetricsUpdate.cgroupMetrics().memoryUsageInBytes()
  235.             self._actual_limit_byte = max(
  236.                 self._actual_limit_byte, executorMetricsUpdate.cgroupMetrics().memoryLimitInBytes()
  237.             )
  239.         task_id = self._executor_task.get(executor_id, None)
  240.         if task_id is None:
  241.             # would still receive metrics updates after job finishes
  242.             self.logger.debug("Cannot find corresponding task for executor {}".format(executor_id))
  243.             return
  245.         self._task_memory[task_id] = max(self._task_memory.get(task_id, 0), memory_byte)
  247.     def getExecutorMetricsFromAPI(self):
  248.         """
  249.         Get Spark metrics from the Spark API
  250.         Potentially switch this to use this end point:
  251.         application_1697883854732_43137/api/v1/applications/application_1697883854732_43137/executors
  252.         Or another option is to record this info in update_stage_info_v3() directly
  253.         """
  254.         self.logger.info("Getting metrics from Spark API")
  255.         app_id = self._sc.applicationId
  256.         proxy_host = self._sc.getConf().get(API_URL_VAR)
  257.         if not proxy_host:
  258.             return
  260.         for stage_id, attempt_id in self._stage_last_attempt_id.items():
  261.             tasks_info_api_url = f"{proxy_host}/api/v1/applications/{app_id}/stages/{stage_id}/{attempt_id}/taskList"
  262.             try:
  263.                 resp = requests.get(tasks_info_api_url)
  264.                 tasks_info_data = resp.json()
  265.                 for task_info in tasks_info_data:
  266.                     task_id = task_info.get("taskId")
  267.                     if task_id is not None:  # warning, this can be 0 (evaluates to False)
  268.                         peak_mem = task_info["taskMetrics"]["peakExecutionMemory"]
  269.                         task_max_mem = max(self._task_memory.get(task_id, 0), peak_mem)
  270.                         self._task_memory[task_info["taskId"]] = task_max_mem
  271.                         self.logger.debug(
  272.                             f"metrics_from_api: stage: {stage_id} task: {task_id} task_max_mem: {task_max_mem}"
  273.                         )
  275.             except Exception as ex:
  276.                 # Even if one fails, we try the others to get some data for the report
  277.                 self.logger.warning(f"metrics_from_api: failed getting tasks_info url: {tasks_info_api_url} ex: {ex}")
  279.     def onJobStart(self, jobStart):
  280.         self._active_jobs += 1
  282.     def onJobEnd(self, jobEnd):
  283.         self._active_jobs -= 1
  284.         job_id = jobEnd.jobId()
  286.         if self._active_jobs == 0:
  287.             self._update_stage_tasks()
  288.             self.logger.info(f"stage_task updated: {self._stage_tasks} job: {job_id}")
  290.             if self._task_memory:
  292.                 self.logger.debug(f"Collected task memory: {self._task_memory} for job_id: {job_id}")
  294.                 self._print_report(self._task_memory)
  296.                 self._save_report(job_id, self._task_memory)
  298.                 self.logger.info("More details at {}".format(get_am_ui_url(self._sc, self._spark_conf)))
  299.             else:
  300.                 if pyspark_v3:
  301.                     # onExecutorMetricsUpdate() isn't working as expected in pyspark 3+, so we use the RESTP API
  302.                     self.logger.info(f"Attempting report via Spark API. version: {pyspark.__version__}")
  303.                     self.getExecutorMetricsFromAPI()
  304.                     self._print_report(self._task_memory)
  305.                     self._save_report(job_id, self._task_memory)
  306.                     self.logger.info("More details at {}".format(get_am_ui_url(self._sc, self._spark_conf)))
  307.                 else:
  308.                     self.logger.info(f"Not enough data collected for the report of job: {job_id}")
  310.             self._reset_memory_stats()
  312.     def onApplicationEnd(self, applicationEnd):
  313.         self.logger.info(
  314.             "Spark application has ended. Access from the history server {}".format(
  315.                 get_history_server_url(self._sc, self._spark_conf)
  316.             )
  317.         )

型

pyspark

2条答案

按热度按时间
kt06eoxx

kt06eoxx1#

问题是我们添加的一个自定义指标没有像之前的spark版本那样可序列化。这导致从worker发送heartbeat时失败。不幸的是，这个问题的唯一提示出现在运行超过120秒的作业时（heartbeat超时）。

赞(0）分享  回复(0）举报  2024-01-06
7xzttuei

7xzttuei2#

我怀疑可能不仅是你的版本改变了，而且你的集群设置也改变了。
executor_id可以等于driver。每个SparkContext中的SparkContext.DRIVER_IDENTIFIER变量等于3.4.1中的driver。每个节点都使用它来确定正在运行的进程是BlockManagerId中的执行器还是驱动程序：

  1. def isDriver: Boolean = {
  2.   executorId == SparkContext.DRIVER_IDENTIFIER
  3. }

字符串
根据您在问题中提供的信息，很难说100%确定（例如，我们不知道您在代码中如何定义executorMetricsUpdate），但如果您的executor_id变量始终等于driver，则可能您正在使用--master local而不是使用实际的集群运行集群。
如果您使用--master local或local的任何变体运行spark应用程序，以便在运行单个JVM进程的单个机器上运行代码，则驱动程序将同时作为驱动程序和执行器执行。
因此，请确保您正在实际的集群上运行，然后您可能会获得想要的行为！

赞(0）分享  回复(0）举报  2024-01-06
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com