<commit_before>package com.couchbase.spark

import com.couchbase.client.java.document.Document
import com.couchbase.spark.connection.{CouchbaseConfig, CouchbaseConnection}
import org.apache.spark.rdd.RDD
import rx.lang.scala.Observable
import rx.lang.scala.JavaConversions._

class DocumentRDDFunctions[D <: Document[_]](rdd: RDD[D]) extends Serializable {

  private val cbConfig = CouchbaseConfig(rdd.context.getConf)

  def saveToCouchbase(bucketName: String = null): Unit = {
    rdd.foreachPartition(iter => {
      val bucket = CouchbaseConnection().bucket(bucketName, cbConfig).async()
      Observable
        .from(iter.toIterable)
        .flatMap(doc => toScalaObservable(bucket.upsert[D](doc)))
        .toBlocking
        .toIterable
        .iterator
    })
  }

}<commit_msg>no need for iterator on save<commit_after>package com.couchbase.spark

import com.couchbase.client.java.document.Document
import com.couchbase.spark.connection.{CouchbaseConfig, CouchbaseConnection}
import org.apache.spark.rdd.RDD
import rx.lang.scala.Observable
import rx.lang.scala.JavaConversions._

class DocumentRDDFunctions[D <: Document[_]](rdd: RDD[D]) extends Serializable {

  private val cbConfig = CouchbaseConfig(rdd.context.getConf)

  def saveToCouchbase(bucketName: String = null) {
    rdd.foreachPartition(iter => {
      val bucket = CouchbaseConnection().bucket(bucketName, cbConfig).async()
      Observable
        .from(iter.toIterable)
        .flatMap(doc => toScalaObservable(bucket.upsert[D](doc)))
        .toBlocking
        .last
    })
  }

}
{'fraction_non_alphanumeric': 0.11324376199616124, 'fraction_numerical': 0.0, 'mean_word_length': 4.077922077922078, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6742381', 'n_tokens_mistral': 510, 'n_tokens_neox': 496, 'n_words': 83}