python - 将 Spark 与 Flask 和 JDBC 一起使用

Question

我在做什么？

我想使用 Flask 构建一个 API 服务，从一个数据库中提取数据，进行一些数据分析，然后将新数据加载到一个单独的数据库中。

怎么了？

如果我自己运行 Spark，我可以访问数据库，执行分析并加载到数据库。但是在 Flask 应用程序(api 路由)中使用它们时，相同的功能将不起作用。

我是怎么做到的？

首先，我启动 Spark master 和 worker。我可以看到我在 master 下的 localhost:8080 中有一名 worker 。

export JAVA_HOME=$(/usr/libexec/java_home -v 1.8)

../sbin/start-master.sh
../sbin/start-slave.sh spark://xxx.local:7077

对于 Flask 应用程序:

app = Flask(__name__)

spark = SparkSession\
    .builder\
    .appName("Flark - Flask on Spark")\
    .getOrCreate()
spark.sparkContext.setLogLevel("ERROR")


@app.route("/")
def hello():
    dataframe = spark.read.format("jdbc").options(
        url="jdbc:postgresql://localhost/foodnome_dev?user=postgres&password=''",
        database="foodnome_test",
        dbtable='"Dishes"'
    ).load()

    print([row["description"]
           for row in dataframe.select('description').collect()])

    return "hello"

为了运行这个应用程序，我使用带有 spark-submit 的 JDBC 驱动程序:

../bin/spark-submit --master spark://Leos-MacBook-Pro.local:7077 --driver-class-path postgresql-42.2.5.jar server.py

我得到了什么错误？

在 Flask 端，错误是内部服务器错误。在 Spark 方面，

File "/Users/leoqiu/Desktop/spark-2.4.0-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
    format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o36.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 3, 10.0.0.67, executor 0): java.lang.ClassNotFoundException: org.postgresql.Driver
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry$.register(DriverRegistry.scala:45)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:55)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:54)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD.compute(JDBCRDD.scala:272)

Answer 1

--driver-class-path 在这里是不够的。驱动程序 jar 也应该添加到执行程序类路径中。这通常使用以下方法一起处理:

• spark.jars.packages/--packages
• spark.jars/--jars

尽管您仍然可以使用 spark.executor.extraClassPath。

解释:

使用 JDBC 源驱动程序负责读取元数据(模式)和实际数据检索过程的执行程序。

这种行为对于不同的外部数据源是常见的，因此无论何时使用非内置格式，都应该在集群中分发相应的 jar。

另见

How to use JDBC source to write and read data in (Py)Spark?