Получить путь к файлу HDFS в PySpark для файлов в формате последовательного файла

Question

Мои данные о HDFS приведены в формате файла последовательности. Я использую PySpark (Спарк 1.6) и пытается достичь 2 вещи:

1. путь данных содержит метку времени в формате гггг/дд/чч/мм, что я хотел бы привести в самих данных. Я попробовал SparkContext.wholeTextFiles, но я думаю, что это может не поддерживать формат файла Sequence.

2. Как мне разобраться с вышеописанной точкой, если я хочу хруст данных в течение дня и хочу ввести дату в данные? В этом случае я бы загружал данные типа yyyy/mm/dd/* format.

Цените любые указатели.

Answer 1

Если сохраненные типы совместимы с типами SQL и вы используете Spark 2.0, это довольно просто. Импорт input_file_name:

from pyspark.sql.functions import input_file_name 

чтения файлов и конвертировать в DataFrame:

df = sc.sequenceFile("/tmp/foo/").toDF() 

Добавить имя файла:

df.withColumn("input", input_file_name()) 

Если это решение не применимо в вашем случае, то универсальный один является (для HDFS вы можете использовать библиотеку hdfs3):

files = ... 

прочитал один за другим, добавляя имя файла:

def read(f): 
    """Just to avoid problems with late binding""" 
    return sc.sequenceFile(f).map(lambda x: (f, x)) 

rdds = [read(f) for f in files] 

и союз:

sc.union(rdds)