Le traitement d'un ResultSet SQL comme un Scala Flux

Quand j'ai une requête à une base de données et de recevoir une (avant uniquement en lecture seule) jeu de résultats, le jeu de résultats agit comme une liste de lignes de la base de données.

Je suis en train d'essayer de trouver un moyen de traiter ce ResultSet comme un Scala Stream. Cela permettra à des opérations telles que le filter, map, etc., alors que ne pas consommer de grandes quantités de RAM.

J'ai mis en place une queue-méthode récursive pour extraire les éléments individuels, mais cela nécessite que tous les éléments soient en mémoire dans le même temps, un problème si le jeu de résultats est très large:

//Iterate through the result set and gather all of the String values into a list
//then return that list
@tailrec
def loop(resultSet: ResultSet,
         accumulator: List[String] = List()): List[String] = {
  if (!resultSet.next) accumulator.reverse
  else {
    val value = resultSet.getString(1)
    loop(resultSet, value +: accumulator)
  }
}
  • Pourriez-vous utiliser un objet iterable au lieu d'un Ruisseau pour faire ce que vous voulez?
  • Également un flux de retenir la valeur dans la mémoire de toute façon si vous ne économiser de la mémoire au moment où vous atteignez la fin de la liste.
  • Je pense que sans jdbc drapeau/option qui rend jdbc lui-même flux, les résultats, vous avez toujours une copie complète des données dans la mémoire, construit par votre api jdbc.
InformationsquelleAutor Ralph | 2012-03-09
  1. 70

    Je n'ai pas tester, mais pourquoi ne serait-il pas de travail?

    new Iterator[String] {
  def hasNext = resultSet.next()
  def next() = resultSet.getString(1)
}.toStream
    • Qui semble parfait. Je vais le tester dès que je reçois ma base de données. Je ne pense pas que j'ai besoin pour le convertir en un Stream. Je peux appliquer map, filter, etc. directement à elle.
    • Je voudrais vous donner un deuxième haut-vote. J'ai ajouté ce fragment de code pour mon Scala extraits de la bibliothèque. Il est rapidement devenu l'un de mes favoris.
    • C'est une bonne solution mais j'ai peur. Je pense que le contrat d'habitude de Iterator est que hasNext est côté-effet-gratuit. Il pourrait être appelé à n'importe quel nombre de fois entre deux appels à next. Est-il quelque chose empêcher que cela devient un problème?
    • Bonne réponse , mais qu'est-ce que la mise en œuvre effective ?
    • Ne fonctionne pas pour moi avec mysql-connector-java version 6. Vous ne savez pas si j'ai fait quelque chose de mal, mais mon ResultSet est fermé sur la deuxième next() appel, afin que je puisse récupérer uniquement une ligne de résultat. La seule façon c'est pas de l'auto-fermé avant j'ai eu toutes les lignes semble être l'utilisation d' while (rs.next()) {...}, j'ai donc ajouter des éléments individuellement à un scala.collection.mutable.ListBuffer dans le while. Ne semble pas assez, mais ne pouvait pas trouver une autre façon.
    • À l'aide de new Iterator[String]{ ... }.toList au lieu de .toStream va chercher l'ensemble des résultats immédiatement, au lieu de simplement la première ligne.

    InformationsquelleAutor elbowich
  2. 10

    Fonction d'utilité pour @elbowich réponse:

    def results[T](resultSet: ResultSet)(f: ResultSet => T) = {
  new Iterator[T] {
    def hasNext = resultSet.next()
    def next() = f(resultSet)
  }
}

    Vous permet d'utiliser l'inférence de type. E. g.:

    stmt.execute("SELECT mystr, myint FROM mytable")

//Example 1:
val it = results(stmt.resultSet) {
  case rs => rs.getString(1) -> 100 * rs.getInt(2)
}
val m = it.toMap //Map[String, Int]

//Example 2:
val it = results(stmt.resultSet)(_.getString(1))
    InformationsquelleAutor hraban
  3. 8

    Cela sonne comme une formidable opportunité pour un implicite de la classe. Tout d'abord, définir l'implicite de la classe quelque part:

    import java.sql.ResultSet

object Implicits {

    implicit class ResultSetStream(resultSet: ResultSet) {

        def toStream: Stream[ResultSet] = {
            new Iterator[ResultSet] {
                def hasNext = resultSet.next()

                def next() = resultSet
            }.toStream
        }
    }
}

    Ensuite, il suffit d'importer l'implicite de la classe partout où vous avez exécuté votre requête et défini l'objet ResultSet:

    import com.company.Implicits._

    Enfin obtenir les données à l'aide de la toStream méthode. Par exemple, obtenir tous les id comme indiqué ci-dessous:

    val allIds = resultSet.toStream.map(result => result.getInt("id"))
    • Êtes-vous sûr qu'il fonctionne? Il échoue sur DB2 avec jeu de résultats en cours de fermeture. Si cela a fonctionné dans votre cas, peut-être que cela dépend de la base de données spécifique de la marque et/ou de configuration?
    • Il n'mais vous ne pouvez utiliser le flux en tant que votre connexion reste ouverte. Si vous fermez votre connexion, le jet échoue, de même que l'itérateur.
    InformationsquelleAutor Jeroen Minnaert
  4. 3

    j'ai besoin de quelque chose de similaire. Bâtiment sur elbowich est très cool réponse, j'ai emballé un peu, et au lieu de la chaîne, j'ai retourner le résultat (de sorte que vous pouvez obtenir n'importe quelle colonne)

    def resultSetItr(resultSet: ResultSet): Stream[ResultSet] = {
    new Iterator[ResultSet] {
      def hasNext = resultSet.next()
      def next() = resultSet
    }.toStream
  }

    J'avais besoin d'accéder à des métadonnées de la table, mais cela fonctionne pour les lignes de la table (on pourrait faire un stmt.executeQuery(sql) au lieu de md.getColumns):

     val md = connection.getMetaData()
 val columnItr = resultSetItr( md.getColumns(null, null, "MyTable", null))
      val columns = columnItr.map(col => {
        val columnType = col.getString("TYPE_NAME")
        val columnName = col.getString("COLUMN_NAME")
        val columnSize = col.getString("COLUMN_SIZE")
        new Column(columnName, columnType, columnSize.toInt, false)
      })
    • Si vous n'avez pas besoin de revenir sur le flux de données (par exemple, avant d'itération uniquement), vous pouvez simplement utiliser un itérateur. Ceci réduit considérablement la charge de la mémoire de l'aide d'un flux (de retour d'une Iterator[ResultSet], et de la baisse de la toStream)
    InformationsquelleAutor Greg
  5. 2

    Parce que ResultSet est juste un objet mutable être navigué par ensuite, nous devons définir notre propre concept de la ligne suivante. Nous pouvons le faire avec une fonction d'entrée comme suit:

    class ResultSetIterator[T](rs: ResultSet, nextRowFunc: ResultSet => T) 
extends Iterator[T] {

  private var nextVal: Option[T] = None

  override def hasNext: Boolean = {
    val ret = rs.next()
    if(ret) {
      nextVal = Some(nextRowFunc(rs))
    } else {
      nextVal = None
    }
    ret
  }

  override def next(): T = nextVal.getOrElse { 
    hasNext 
    nextVal.getOrElse( throw new ResultSetIteratorOutOfBoundsException 
  )}

  class ResultSetIteratorOutOfBoundsException extends Exception("ResultSetIterator reached end of list and next can no longer be called. hasNext should return false.")
}

    EDIT:
    Traduire de diffuser ou de quelque chose d'autre que ci-dessus.

    InformationsquelleAutor Brendan
  6. 0

    Cette mise en œuvre, bien que plus long et plus encombrant il est dans une meilleure correspondance avec le jeu de résultats du contrat. L'effet secondaire n'a été supprimé depuis hasNext(...) et a déménagé dans next(). 

    new Iterator[String] {
  private var available = resultSet.next()
  override def hasNext: Boolean = available
  override def next(): String = {
    val string = resultSet.getString(1)
    available = resultSet.next()
    string
  }
}
    InformationsquelleAutor thoredge
  7. 0

    Je pense que la plupart des au-dessus des implémentations a un non déterministe hasNext méthode. De l'appeler deux fois permet de déplacer le curseur à la deuxième ligne. Je vous conseille d'utiliser quelque chose comme ça:

      new Iterator[ResultSet] {
    def hasNext = {
      !resultSet.isLast
    }
    def next() = {
      resultSet.next()
      resultSet
    }
  }
    InformationsquelleAutor Matzz
  8. 0
    Iterator.continually(rs.next())
  .takeWhile(identity)
  .map(_ => Model(
      id = rs.getInt("id"),
      text = rs.getString("text")
   ))
    InformationsquelleAutor Sergey Alaev