Simple script showing some operations with Apache Flink and Scala.

flink_batch.scala

package org.example

import org.apache.flink.api.scala._

object Pictures {
  def main(args: Array[String]) {
    // set up the execution environment
    val env = ExecutionEnvironment.getExecutionEnvironment

    case class Movie
    (
      name: String,
      year: String,
      nominations: String,
      rating: String,
      duration: String,
      genre1: String,
      genre2: String,
      release: String,
      metacritic: String,
      synopsis: String
    )

    val text_path = "/Users/adrian/Documents/proyectos/flink-project/src/main/resources/pictures.csv"
    val mydataset: DataSet[Movie] = env.readCsvFile[Movie](text_path, ignoreFirstLine = true)

    val filtered_movies = mydataset
      .filter(
        movie => movie.nominations != "-" && movie.metacritic != "-" && movie.metacritic != ""
      )

    val nominations_sum = filtered_movies
      .map(movie => movie.nominations.toInt).reduce(_+_)


    val nominations_count: Float = filtered_movies
      .map(movie => movie.nominations.toInt).count().toFloat

    val sum: Int = nominations_sum.collect().head
    val count = nominations_count
    val nominations_avg = sum / count

    println(nominations_avg)


    val grouped_by_genres = filtered_movies.groupBy("genre1")
      .reduceGroup {
        movies => {
          val movies_list = movies.toList
          val size = movies_list.length
          val total: Int = movies_list.foldLeft(0)((acc, element) => acc + element.metacritic.toInt)

          (movies_list.head.genre1, total / size.toFloat)
        }
      }

    grouped_by_genres.print()

    // execute program
    env.execute("Pictures")
  }
}