Spark - получение имени файла в RDD

Ответ 1

Начиная с Spark 1.6 вы можете комбинировать источник данных text и input_file_name следующим образом:

Scala

import org.apache.spark.sql.functions.input_file_name

val inputPath: String = ???

spark.read.text(inputPath)
  .select(input_file_name, $"value")
  .as[(String, String)] // Optionally convert to Dataset
  .rdd // or RDD

Python

(Версии до 2.x являются ошибками и не могут сохранять имена при преобразовании в RDD):

from pyspark.sql.functions import input_file_name

(spark.read.text(input_path)
    .select(input_file_name(), "value"))
    .rdd)

Это можно использовать и с другими форматами ввода.

Ответ 2

Если ваши текстовые файлы достаточно малы, вы можете использовать SparkContext.wholeTextFiles, который возвращает RDD (filename,content).

Ответ 3

Если ваши текстовые файлы слишком велики для SparkContext.wholeTextFiles, вы должны использовать (простой) пользовательский InputFormat, а затем вызвать SparkContext.hadoopRDD

В InputFormat необходимо вернуть кортеж (имя файла, строка), а не строку, тогда вы можете фильтровать с использованием предиката, который просматривает содержимое строки, затем уникально и собирает имена файлов.

Из Spark код будет выглядеть примерно так:

val ft = classOf[FileNamerInputFormat]
val kt = classOf[String]
val vt = classOf[String]

val hadoopConfig = new Configuration(sc.hadoopConfiguration)
sc.newAPIHadoopFile(path, ft, kt, vt, hadoopConfig)
  .filter { case (f, l) => isInteresting(l) }
  .map { case (f, _) => f } 
  .distinct()
  .collect()

Ответ 4

Вы можете попробовать это, если находитесь в pyspark:

    test = sc.wholeTextFiles("pathtofile")

вы получите результирующий RDD с первым элементом = путь к файлу, а второй элемент = содержимое

Ответ 5

Вы можете использовать WholeTextFile() для достижения этого. Однако, если входные файлы большие, тогда было бы встречно работать с WholeTextFile(), поскольку оно помещало весь контент файла в одну запись.

Лучший способ получить имена файлов в таком сценарии - использовать mapPartitionsWithInputSplit(). Вы можете найти рабочий пример, используя этот сценарий мой блог.

Ответ 6

Кажется, излишне использовать Spark напрямую... Если эти данные будут "собраны" водителю, почему бы не использовать API HDFS? Часто Hadoop поставляется вместе с Spark. Вот пример:

import org.apache.hadoop.fs._
import org.apache.hadoop.conf._

val fileSpec = "/data/Invoices/20171123/21"
val conf = new Configuration()
val fs = org.apache.hadoop.fs.FileSystem.get(new URI("hdfs://nameNodeEneteredHere"),conf)
val path = new Path(fileSpec)
// if(fs.exists(path) && fs.isDirectory(path) == true) ...
val fileList = fs.listStatus(path)

Затем с println(fileList(0)) информация (форматированная), подобная этому первому элементу (в качестве примера), можно увидеть как org.apache.hadoop.fs.FileStatus:

FileStatus {
    path=hdfs://nameNodeEneteredHere/Invoices-0001.avro; 
    isDirectory=false; 
    length=29665563;
    replication=3;
    blocksize=134217728;
    modification_time=1511810355666;
    access_time=1511838291440;
    owner=codeaperature;
    group=supergroup;
    permission=rw-r--r--;
    isSymlink=false
}

Где fileList(0).getPath даст hdfs://nameNodeEneteredHere/Invoices-0001.avro.

Я предполагаю, что это означает, что чтение файлов будет в первую очередь с помощью namenode HDFS, а не внутри каждого исполнителя. TL;DR; Я уверен, что Spark, скорее всего, опросит namenode, чтобы получить RDD. Если базовый вызов Spark опросит namenode для управления RDD, возможно, вышеописанное является эффективным решением. Тем не менее, содержательные комментарии, предлагающие любое направление, будут приветствоваться.

Ответ 7

Если вы используете Dataframe API, вы можете получить имена файлов из HDFS, используя функцию input_file_name из org.apache.spark.sql.functions. Приведенные ниже фрагменты могут помочь вам понять.

val df = spark.read.csv("/files/")
val df2 = df.withColumn("file_name", split(input_file_name(), "/").getItem(7).cast(StringType)) 
val df3 = df.withColumn("file_name", input_file_name()) 

df2 теперь включает новое поле с именем "имя_файла", которое содержит имя файла HDFS, извлеченное с помощью функции split. Если вам нужен полный путь HDFS, вы можете использовать функцию input_file_name() только как показано в df3.