python - 流式 pyspark 应用程序中的连接池

Question

​在流式 pyspark 应用程序中使用连接池的正确方法是什么？

我通读了https://forums.databricks.com/questions/3057/how-to-reuse-database-session-object-created-in-fo.html并理解正确的方法是对 scala/java 使用单例。这在 python 中可能吗？一个小的代码示例将不胜感激。我相信为每个分区创建连接对于流应用程序来说效率非常低。

Answer 1

由于 PySpark 架构的原因，长话短说，连接池在 Python 中的用处不如在 JVM 中有用。与 Scala 对应的 Python 执行器不同，Python 执行器使用单独的进程。这意味着执行器之间没有共享状态，并且由于默认情况下每个分区都是按顺序处理的，因此每个解释器只能有一个事件连接。

当然，维持批处理之间的连接仍然很有用。要实现这一目标，您需要做两件事:

• spark.python.worker.reuse 必须设置为 true。
• 一种在不同调用之间引用对象的方法。

第一个非常明显，第二个并不是 Spark 特有的。例如，您可以使用模块单例(您将在我对 How to run a function on all Spark workers before processing data in PySpark? 的回答中找到 Spark 示例)或 Borg pattern .