title: Flink - Why should I create my own RichSinkFunction instead of just open and close my PostgreSql connection?
body: <p>I would like to know why I really need to create my own RichSinkFunction or use JDBCOutputFormat to connect on the database instead of just Create my connection, perform the query and close the connection using the traditional PostgreSQL drivers inside my SinkFunction?</p>
  
  <p>I found many articles telling do to that but does not explain why? What is the difference?</p>
  
  <p>Code example using JDBCOutputFormat, </p>
  
  <pre><code>JDBCOutputFormat jdbcOutput = JDBCOutputFormat.buildJDBCOutputFormat()
       .setDrivername("org.postgresql.Driver")
       .setDBUrl("jdbc:postgresql://localhost:1234/test?user=xxx&amp;password=xxx")
       .setQuery(query)
       .setSqlTypes(new int[] { Types.VARCHAR, Types.VARCHAR, Types.VARCHAR }) //set the types
       .finish();
  </code></pre>
  
  <p>Code example implementing the own RichSinkFunction,</p>
  
  <pre><code>public class RichCaseSink extends RichSinkFunction&lt;Case&gt; {
  
    private static final String UPSERT_CASE = "INSERT INTO public.cases (caseid, tracehash) "
        + "VALUES (?, ?) "
        + "ON CONFLICT (caseid) DO UPDATE SET "
        + "  tracehash=?";
  
    private PreparedStatement statement;
  
  
    @Override
    public void invoke(Case aCase) throws Exception {
  
      statement.setString(1, aCase.getId());
      statement.setString(2, aCase.getTraceHash());
      statement.setString(3, aCase.getTraceHash());
      statement.addBatch();
      statement.executeBatch();
    }
  
    @Override
    public void open(Configuration parameters) throws Exception {
      Class.forName("org.postgresql.Driver");
      Connection connection =
          DriverManager.getConnection("jdbc:postgresql://localhost:5432/casedb?user=signavio&amp;password=signavio");
  
      statement = connection.prepareStatement(UPSERT_CASE);
    }
  
  }
  </code></pre>
  
  <p>why I cannot just use the PostgreSQL driver?</p>
  
  <pre><code>public class Storable implements SinkFunction&lt;Activity&gt;{
  
      @Override
      public void invoke(Activity activity) throws Exception {
          Class.forName("org.postgresql.Driver");
          try(Connection connection =
              DriverManager.getConnection("jdbc:postgresql://localhost:5432/casedb?user=signavio&amp;password=signavio")){
  
          statement = connection.prepareStatement(UPSERT_CASE);
  
          //Perform the query
  
          //close connection...
          }
      }
  
  }
  </code></pre>
  
  <p>Does someone know the technical answer to the best practice in Flink? 
  Does Implementation of RichSinkFunction or usage of JDBCOutputFormat do something special?</p>
  
  <p>Thank you in advance.</p>