Перенос имени каталога файла журнала в фрейм данных Pyspark

Question

У меня немного странный. У меня есть куча журналов, которые мне нужно тралить. Я успешно сделал это в Spark, и я доволен этим.

Однако мне нужно добавить еще одно поле в фрейм данных, который является центром обработки данных.

Единственное место, гдеИмя центра данных может быть получено из пути к каталогу.

Например:

/feedname/date/datacenter/another/logfile.txt

Каким образом можно извлечь путь к файлу журнала и вставить его в кадр данных? Оттуда я могу сделать несколько строк и извлечь нужный мне бит.

Мой текущий код:

mpe_data = my_spark.read\
    .option("header","false")\
    .option("delimiter", "\t")\
    .withColumn("Datacenter", input_file_name())\
    .csv('hdfs://nameservice/data/feed/mpe/dt=20191013/*/*/*', final_structure)
mpe_data.printSchema()
mpe_data.createOrReplaceTempView("mpe")

Answer 1

Вы можете получить путь к файлу, используя _input_file_name_ в Spark 2.0 +

from pyspark.sql.functions import input_file_name 
df.withColumn("Datacenter", input_file_name())

Добавляя свой фрагмент кода в качестве примера, как только вы прочитали свой файл, используйте withcolumn для получения имени файла.

mpe_data = my_spark.read\
    .option("header","false")\
    .option("delimiter", "\t")\
    .csv('hdfs://nameservice/data/feed/mpe/dt=20191013/*/*/*', final_structure)

mpe_data.withColumn("Datacenter", input_file_name())

mpe_data.printSchema()