如何重新运行ApacheFlinkPostgresJDBC作业而不出现“找不到合适的驱动程序”异常

Question

我有一份来自starter Maven项目的Flink工作。该作业有一个打开Postgres JDBC连接的源。我正在我自己的Flink会话集群上使用示例docker compose执行作业。yml。

当我第一次提交作业时，它会成功执行。当我尝试再次提交时，我得到以下错误：

Caused by: java.sql.SQLException: No suitable driver found for jdbc:postgresql://host.docker.internal:5432/postgres?user=postgres&password=mypassword
    at java.sql.DriverManager.getConnection(DriverManager.java:689)
    at java.sql.DriverManager.getConnection(DriverManager.java:270)
    at com.myorg.project.JdbcPollingSource.run(JdbcPollingSource.java:25)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)

我必须重新启动我的集群，以便重新运行我的工作。为什么会发生这种情况？如何在不重新启动群集的情况下再次提交作业？

Maven入门项目的唯一补充是：

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.2.24</version>
</dependency>

Flink源代码只打开JDBC连接，如下所示：

package com.mycompany;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;

public class JdbcSource extends RichSourceFunction<Integer> {

    private final String connString;

    public JdbcSource(String connString) {
        this.connString = connString;
    }

    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
        try (Connection conn = DriverManager.getConnection(this.connString)) {

        }
    }

    @Override
    public void cancel() {
    }
}

我已经在Flink版本1.14.0和1.13.2上测试了这个，结果相同。

请注意，这个问题提供了在我的RichSourceFunction中使用Class.forName（"org.postgresql.驱动程序"）；的解决方案。然而，我想知道发生了什么事。

Answer 1

根据PostgreSQL JDBC驱动程序的官方文档，如果您使用的是Java 1.6，您可以将驱动程序的jar文件放入类路径中。JVM将自动加载驱动程序。所以问题是如何将驱动程序的jar文件放入类路径中。

由于您使用docker部署会话群集，因此有两种方法可以工作：

1. 将驱动程序的jar文件放入docker映像

使用以下命令运行并访问图像：

docker docker run -it -v $PWD:/tmp/flink <address to image> -- bash

将驱动程序的jar文件复制到文件夹/opt/flink/lib。

从容器创建新映像。由于默认情况下/pt/flink/lib作为类路径加载，现在驱动程序的jar文件位于类路径。

将maven汇编插件添加到pom。maven项目的xml。重新编译您的项目并获得一个带有依赖项的jar文件。在这个jar中，PostgreSQL JDBC驱动程序打包在一起。

Answer 2

我对Apache Flink 1.13的Postgres有pom.xml依赖：

        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <version>9.4-1201-jdbc41</version>
        </dependency>

您可以使用Postgres连接器类，例如：

public class PostgreSQLConnector {
    private static volatile PostgreSQLConnector instance;
    private Connection connectionDB = null;

    public PostgreSQLConnector(your params) {
        ...
    }

    public static PostgreSQLConnector getInstance() {
        PostgreSQLConnector postgreSQLConnector = instance;
        if (postgreSQLConnector != null)
            return postgreSQLConnector;
        synchronized (PostgreSQLConnector.class) {
            if (instance == null) {
                instance = new PostgreSQLConnector(your params);
            }
            return instance;
        }
    }

    public Connection getConnectionDB() throws SQLException {
        if (checkNullConnection()) CreateConnection();
        return connectionDB;
    }

    public void CheckConnection() throws SQLException {
        if (checkNullConnection()) CreateConnection();
    }

    public void CreateConnection() throws SQLException {
            try {
                Class.forName(sink.driverName);
                connectionDB = DriverManager.getConnection(fullUrl, username, password);
            } catch (Exception e) {
                ...
            }
    }

public boolean checkNullConnection() throws SQLException {
        return (connectionDB == null || connectionDB.isClosed());
    }
}

然后，您可以创建一个RichSourceFunction，并在重写open方法中创建连接，而不是在run中创建连接。

public class JdbcSource extends RichSourceFunction<Integer> {

    private final String connString;
    private static Connection dbConnection;
    private static final PostgreSQLConnector postgreSQLConnector = PostgreSQLConnector.getInstance();

    public JdbcSource(String connString) {
        this.connString = connString;
    }

    @Override
    public void open(Configuration parameters) throws SQLException {
        dbConnection = postgreSQLConnector.getConnectionDB();
    }

    @Override
    public void close() throws Exception {
        if (dbConnection != null) dbConnection.close();
    }

    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
       do something here with the connection
    }

    @Override
    public void cancel() {
    }
}

像这样的东西你可以试试，它应该会起作用

Answer 3

从ApacheFlink中的SQL数据库中读取数据集时，找不到JDBC驱动程序的第一个问题。

第二，如果使用会话模式。可以很容易地重新运行Flink作业，而无需重新启动集群。您可以登录作业管理器shell，然后使用命令rerun job。

类。forName（“org.postgresql.Driver”）将触发静态方法块，所以DriverManager可以获取驱动程序类。见：

// from org.postgresql.Driver
    static {
        try {
            register();
        } catch (SQLException var1) {
            throw new ExceptionInInitializerError(var1);
        }
    }