浅谈Azure数据库的多处理

Question

最近，我的任务是在Databricks Delta-lake上摄取JSON响应。我必须使用不同的参数访问REST端点URL 6500次，并提取响应。

我尝试了多处理库中的两个模块，ThreadPool和Pool，以使每个模块的执行速度更快一些。

ThreadPool:

1. 当集群被设置为从2个工作节点到13个工作节点时，如何选择ThreadPool的线程数？

现在，我已经设置了n_pool = multiprocessing.cpu_count()，如果集群自动缩放，会有什么不同吗？

池

1. 当我使用池来使用处理器而不是线程时。我在每次执行时都会随机地看到以下错误。好吧，我从Session/Conf丢失的错误中了解到，我需要从每个进程中设置它。但是，我在Databricks上，启用了默认星火会话，那么为什么我会看到这些错误。

Py4JError: SparkConf does not exist in the JVM 
**OR** 
py4j.protocol.Py4JError: org.apache.spark.api.python.PythonUtils.getEncryptionEnabled does not exist in the JVM

1. 最后，计划将多进程替换为“concurrent.futures.ProcessPoolExecutor”。有什么区别吗？

Answer 1

如果使用线程池，它们将只在驱动程序节点上运行，执行器将处于空闲状态。相反，您需要使用Spark本身来并行化请求。这通常是通过创建一个带有URL列表(如果基本URL是相同的URL参数)的数据same来完成的，然后使用Spark来执行实际的请求。就像这样：

import urllib

df = spark.createDataFrame([("url1", "params1"), ("url2", "params2")], 
                           ("url", "params"))

@udf("body string, status int")
def do_request(url: str, params: str):
  full_url = url + "?" + params # adjust this as required
  with urllib.request.urlopen(full_url) as f:
    status = f.status
    body = f.read().decode("utf-8")
  
  return {'status': status, 'body': body}
  

res = df.withColumn("result", do_requests(col("url"), col("params")))

这将使用一个名为result的新列返回dataframe，该列将有两个字段-- status和body (JSON答案作为字符串)。

Answer 2

您可以尝试以下方法来解决

Py4JError: SparkConf does not exist in the JVM
**OR** 
py4j.protocol.Py4JError: org.apache.spark.api.python.PythonUtils.getEncryptionEnabled does not exist in the JVM

错误

Install findspark 
$pip install findspark
Code:
import findsparkfindspark.init()

参考文献：Py4JError: JVM中不存在SparkConf和org.apache.spark.api.python.PythonUtils.getEncryptionEnabled : JVM中不存在org.apache.spark.api.python.PythonUtils.getEncryptionEnabled