quill-spark.scala

def topHashtags(tweets: Dataset[Tweet], n: Int): Dataset[(String, Long)] =
  run {                               // produce a dataset from the Quill query
    liftQuery(tweets)                 // trasform the dataset into a Quill query
      .concatMap(_.text.split(" "))   // split into words and unnest results
      .filter(_.startsWith("#"))      // filter hashtag words
      .map(_.toLowerCase)             // normalize hashtags
      .groupBy(word => word)          // group by each hashtag
      .map {                          // map word list to its count
        case (word, list) => 
          (word, list.size)
      }
      .sortBy {                       // sort by the count desc
        case (word, count) => -count
      }
      .take(lift(n))                  // limit to top results
  }