<commit_before>package uk.ac.wellcome.platform.snapshot_generator.source

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Sink, Source}
import com.sksamuel.elastic4s.http.ElasticDsl.{search, termQuery}
import com.sksamuel.elastic4s.http.HttpClient
import com.sksamuel.elastic4s.http.search.SearchHit
import com.sksamuel.elastic4s.streams.ReactiveElastic._
import com.twitter.inject.Logging
import uk.ac.wellcome.models.IdentifiedWork
import uk.ac.wellcome.utils.JsonUtil._

object ElasticsearchWorksSource extends Logging {
  def apply(elasticClient: HttpClient, indexName: String, itemType: String)(
    implicit actorSystem: ActorSystem): Source[IdentifiedWork, NotUsed] = {
    val loggingSink = Flow[IdentifiedWork]
      .grouped(10000)
      .map(works => {
        logger.info(s"Received ${works.length} works from $indexName")
        works
      })
      .to(Sink.ignore)
    Source
      .fromPublisher(
        elasticClient.publisher(
          search(s"$indexName/$itemType")
            .query(termQuery("visible", true))
            .scroll(keepAlive = "2m")
            .size(1000))
      )
      .map { searchHit: SearchHit =>
        fromJson[IdentifiedWork](searchHit.sourceAsString).get
      }
      .alsoTo(loggingSink)
  }
}
<commit_msg>Comment<commit_after>package uk.ac.wellcome.platform.snapshot_generator.source

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Sink, Source}
import com.sksamuel.elastic4s.http.ElasticDsl.{search, termQuery}
import com.sksamuel.elastic4s.http.HttpClient
import com.sksamuel.elastic4s.http.search.SearchHit
import com.sksamuel.elastic4s.streams.ReactiveElastic._
import com.twitter.inject.Logging
import uk.ac.wellcome.models.IdentifiedWork
import uk.ac.wellcome.utils.JsonUtil._

object ElasticsearchWorksSource extends Logging {
  def apply(elasticClient: HttpClient, indexName: String, itemType: String)(
    implicit actorSystem: ActorSystem): Source[IdentifiedWork, NotUsed] = {
    val loggingSink = Flow[IdentifiedWork]
      .grouped(10000)
      .map(works => {
        logger.info(s"Received ${works.length} works from $indexName")
        works
      })
      .to(Sink.ignore)
    Source
      .fromPublisher(
        elasticClient.publisher(
          search(s"$indexName/$itemType")
            .query(termQuery("visible", true))
            .scroll(keepAlive = "2m")
            // Increasing the size of each request from the
            // default 100 to 1000 as it makes it go significantly faster
            .size(1000))
      )
      .map { searchHit: SearchHit =>
        fromJson[IdentifiedWork](searchHit.sourceAsString).get
      }
      .alsoTo(loggingSink)
  }
}

{'fraction_non_alphanumeric': 0.106539309331374, 'fraction_numerical': 0.012858192505510653, 'mean_word_length': 3.3359872611464967, 'pattern_counts': {'":': 0, '<': 3, '<?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': '383092', 'n_tokens_mistral': 872, 'n_tokens_neox': 849, 'n_words': 153}