Scala / Spark - Найти общее количество значений в строке на основе ключа

Question

У меня большой текстовый файл, который содержит просмотры страниц некоторых проектов Викимедиа. (Вы можете найти его здесь , если вы действительно заинтересованы). Каждая строка, разделенная пробелом, содержит статистику для одной страницы Викимедиа. Схема выглядит следующим образом: <project code> <page title> <num hits> <page size>

В Scala, используя Spark RDD или Dataframes, I w sh для вычисления общего количества обращений для каждого проекта на основе кода проекта. Так, например, для проектов с кодом «zw» я хотел бы найти все строки, начинающиеся с кода проекта «zw», и сложить их совпадения. Очевидно, это должно быть сделано для всех кодов проекта одновременно.

Я рассмотрел такие функции, как aggregateByKey et c, но примеры, которые я обнаружил, не go достаточно подробно, особенно для файла с 4 поля. Я полагаю, что это какая-то работа MapReduce, но как ее реализовать, мне не под силу.

Любая помощь будет принята с благодарностью.

Answer 1

Во-первых, вы должны прочитать файл как Dataset[String]. Затем разберите каждую строку в кортеж, чтобы ее можно было легко преобразовать в Dataframe. Если у вас есть Dataframe, простого .GroupBy().agg() достаточно, чтобы завершить вычисление sh.

import org.apache.spark.sql.functions.sum

val df = spark.read.textFile("/tmp/pagecounts.gz").map(l => {
    val a = l.split(" ")
    (a(0), a(2).toLong)
}).toDF("project_code", "num_hits")

val agg_df = df.groupBy("project_code")
  .agg(sum("num_hits").as("total_hits"))
  .orderBy($"total_hits".desc)

agg_df.show(10)

В приведенном выше фрагменте показаны 10 лучших кодов проекта по общему количеству посещений.

+------------+----------+
|project_code|total_hits|
+------------+----------+
|       en.mw|   5466346|
|          en|   5310694|
|       es.mw|    695531|
|       ja.mw|    611443|
|       de.mw|    572119|
|       fr.mw|    536978|
|       ru.mw|    466742|
|          ru|    463437|
|          es|    400632|
|       it.mw|    400297|
+------------+----------+

Конечно, также возможно сделать это с более старым API в виде RDD сопоставления / уменьшения, но вы потеряете многие оптимизации, которые приносит Dataset / Dataframe api.