Risultato SQL SQL a InputStream

Question

Ho bisogno di una funzione Java che restituisca i risultati di una query SQL SELECT come parametro InputStream per un altro sistema che invia il risultato su una rete.

Tuttavia, lo InputStream deve essere di un String con delimitatori personalizzati (ad esempio, spesso, ma non sempre, CSV).

Mentre posso facilmente creare una funzione per recuperare il risultato, creare una delimitato String, e infine convertire che String a un InputStream, il risultato SQL sarà spesso troppo grande per elaborare in memoria. Inoltre, l'elaborazione dell'intero set di risultati prima di restituire il risultato incorrerà in un tempo di attesa indesiderato.

Come posso restituire un InputStream per eseguire un'iterazione sul risultato SQL e inviare i dati elaborati (delimitati) come vengono restituiti dal database?

Answer 1

distacco (non testato) frammento di codice, che dovrebbe darvi un'idea di base:

/** 
* Implementors of this interface should only convert current row to byte array and return it. 
* 
* @author yura 
*/ 
public interface RowToByteArrayConverter { 
    byte[] rowToByteArray(ResultSet resultSet); 
} 

public class ResultSetAsInputStream extends InputStream { 

    private final RowToByteArrayConverter converter; 
    private final PreparedStatement statement; 
    private final ResultSet resultSet; 

    private byte[] buffer; 
    private int position; 

    public ResultSetAsInputStream(final RowToByteArrayConverter converter, final Connection connection, final String sql, final Object... parameters) throws SQLException { 
     this.converter = converter; 
     statement = createStatement(connection, sql, parameters); 
     resultSet = statement.executeQuery(); 
    } 

    private static PreparedStatement createStatement(final Connection connection, final String sql, final Object[] parameters) { 
     // PreparedStatement should be created here from passed connection, sql and parameters 
     return null; 
    } 

    @Override 
    public int read() throws IOException { 
     try { 
      if(buffer == null) { 
       // first call of read method 
       if(!resultSet.next()) { 
        return -1; // no rows - empty input stream 
       } else { 
        buffer = converter.rowToByteArray(resultSet); 
        position = 0; 
        return buffer[position++] & (0xff); 
       } 
      } else { 
       // not first call of read method 
       if(position < buffer.length) { 
        // buffer already has some data in, which hasn't been read yet - returning it 
        return buffer[position++] & (0xff); 
       } else { 
        // all data from buffer was read - checking whether there is next row and re-filling buffer 
        if(!resultSet.next()) { 
         return -1; // the buffer was read to the end and there is no rows - end of input stream 
        } else { 
         // there is next row - converting it to byte array and re-filling buffer 
         buffer = converter.rowToByteArray(resultSet); 
         position = 0; 
         return buffer[position++] & (0xff); 
        } 
       } 
      } 
     } catch(final SQLException ex) { 
      throw new IOException(ex); 
     } 
    } 



    @Override 
    public void close() throws IOException { 
     try { 
      statement.close(); 
     } catch(final SQLException ex) { 
      throw new IOException(ex); 
     } 
    } 
} 

Questa è l'implementazione molto diretto e può essere migliorato in modi seguenti:

• codice la duplicazione tra if e else nel metodo read può essere rimossa - è stata pubblicata solo per chiarimenti
• invece di ricreare il buffer di array di byte per ogni riga (new byte[] è un'opera costosa zione), è possibile implementare una logica più sofisticata per utilizzare il buffer di array di byte che viene inizializzato solo una volta e quindi riempito nuovamente. Uno quindi dovrebbe cambiare la firma del metodo RowToByteArrayConverter.rowToByteArray su int fillByteArrayFromRow(ResultSet rs, byte[] array) che dovrebbe restituire il numero di byte riempiti e riempire l'array di byte passato.

Poiché byte contiene byte firmato può contenere -1 (che è in realtà 255 byte senza segno) e quindi indicare fine errata di flusso, così & (0xff) viene utilizzato per convertire byte con segno a byte senza segno come valori interi. Per dettagli, fare riferimento a How does Java convert int into byte?.

Si noti inoltre che se la velocità di trasferimento di rete è lenta, è possibile che i set di risultati aperti per rimangano a lungo aperti, ponendo problemi al database.

Spero che questo aiuti ...

Answer 2

che vorrei migliorare la risposta suggerita da @Yura, introducendo il seguente:
Usa DataOutputStream che viene inizializzato con un ByteArrayOutputStream per scrivere comodamente i dati per l'array di byte, all'interno di un'implementazione di RowToByteArrayConverter.
In realtà, vorrei suggerire di avere una gerarchia di convertitori, tutti loro estendere la stessa classe astratta (si tratta di un frammento di codice della mia idea - potrebbe non compilare dai prima volta)

public abstract class RowToByteArrayConverter { 
    public byte[] rowToByteArray(ResultSet resultSet) { 
     parseResultSet(dataOutputStream, resultSet); 
     return byteArrayOutputSteam.toByteArray(); 
    } 

    public RowToByteArrayConverter() { 
    dataOutputStream = new DataOutputStream(byteArrayOutputStream); 
    } 

    protected DataOutputStream dataOutputStream; 
    protected ByteArrayOutputStream byteArrayOutputStream; 

    protected abstract void parseResultSet(DataOutputStream dataOutputStresm, ResultSet rs); 
} 

Ora, può sovrascrivere questa classe semplicemente sovrascrivendo il metodo parseResultSet, ad esempio
- codice di scrittura che ottiene come stringa un nome da una colonna "nome" nel record. ed esegue writeUTF8 su DataOputputStream.

Answer 3

Le risposte di cui sopra forniscono una soluzione utile al problema di un stringbuilder di dimensioni limitate superato.Sono anche efficienti in termini di memoria. Tuttavia, il mio test suggerisce che essi sono più lenti di una semplice scrittura di dati in uno StringBuilder, e chiamando

nuovo ByteArrayInputStream (data.getBytes ("UTF-8"))

per ottenere un InputStream.

Quello che ho trovato per essere di gran lunga più performante è quello di tagliare i dati in arrivo utilizzando una funzione di partizione e quindi utilizzando più thread per ciascuna:

1. query al database di origine per un sottoinsieme dei dati
2. Scrivi i dati sul target

Questo evita anche il problema in cui i dati totali possono superare la dimensione massima di un buffer di stringa.

Ad esempio, ho record di 6 m con una colonna denominata "RecordDate" in una tabella di SQL Server. I valori in Recorddate variano tra il 2013 e il 2016. Pertanto, configuro ciascun thread per ogni richiesta i dati per 2013,14,15,16, rispettivamente. Quindi ogni thread scrive i dati transcodificati su uno StringBuilder e ciascun carico di massa verso la destinazione convertendosi in un Inputstream usando getBytes() come sopra.

Questo ha comportato un aumento di velocità 2x.

Perché? Poiché i database di origine e di destinazione sono in grado di gestire più richieste simultanee e pertanto il carico di lavoro complessivo viene distribuito su più thread in tutti e tre i processi: database di origine, transcoder, database di destinazione.