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

Question

Я пытаюсь обрабатывать 4 каталога текстовых файлов, которые постоянно растут каждый день. Что мне нужно сделать, если кто-то пытается найти номер счета-фактуры, я должен предоставить им список файлов.

Мне удалось сопоставить и уменьшить значения в текстовых файлах, загрузив их как RDD. Но как я могу получить имя файла и другие атрибуты файла?

Answer 1

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

Answer 2

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

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

От искры, код будет выглядеть примерно так:

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() 

Answer 3

С Спарк 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.х глючат и не может сохранить имена при преобразовании в RDD):

from pyspark.sql.functions import input_file_name 

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

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

Answer 4

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

test = sc.wholeTextFiles("pathtofile") 

вы получите в результате RDD с первым элементом = FilePath и второй элемент = содержимому

Answer 5

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

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

Answer 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.

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