如何从udf调用文件系统

Question

目标是为每个DataFrame行添加具有修改时间的列。

鉴于

val data = spark.read.parquet("path").withColumn("input_file_name", input_file_name())

+----+------------------------+
| id |        input_file_name |
+----+------------------------+
|  1 | hdfs://path/part-00001 |
|  2 | hdfs://path/part-00001 |
|  3 | hdfs://path/part-00002 |
+----+------------------------+

预期的

+----+------------------------+
| id |      modification_time |
+----+------------------------+
|  1 | 2000-01-01Z00:00+00:00 |
|  2 | 2000-01-01Z00:00+00:00 |
|  3 | 2000-01-02Z00:00+00:00 |
+----+------------------------+

我写了一个函数来获取修改时间

def getModificationTime(path: String): Long = {
    FileSystem.get(spark.sparkContext.hadoopConfiguration)
        .getFileStatus(new org.apache.hadoop.fs.Path(path))
        .getModificationTime()
}

val modificationTime = getModificationTime("hdfs://srsdev/projects/khajiit/data/OfdCheques2/date=2020.02.01/part-00002-04b9e4c8-5916-4bb2-b9ff-757f843a0142.c000.snappy.parquet")

修改时间：长=1580708401253

...但它在查询中不起作用

def input_file_modification_time = udf((path: String) => getModificationTime(path))

data.select(input_file_modification_time($"input_file_name") as "modification_time").show(20, false)

组织。阿帕奇。火花SparkException:作业因阶段失败而中止：阶段54.0中的任务0失败4次，最近的失败：阶段54.0中的任务0.3丢失（TID 408，srs-hdp-s1.dev.kontur.ru，executor 3）：org。阿帕奇。火花SparkException:无法执行用户定义的函数（$anonfun$input_file_MODIFY_time$1:（字符串）=

Answer 1

注意为数据帧的每一行调用getModificationTime将对性能产生影响。

修改代码以一次性获取文件元数据

请检查下面的代码。

scala> val df = spark.read.format("parquet").load("/tmp/par")
df: org.apache.spark.sql.DataFrame = [id: int]

scala> :paste
// Entering paste mode (ctrl-D to finish)

def getModificationTime(path: String): Long = {
    FileSystem.get(spark.sparkContext.hadoopConfiguration)
        .getFileStatus(new org.apache.hadoop.fs.Path(path))
        .getModificationTime()
}

// Exiting paste mode, now interpreting.

getModificationTime: (path: String)Long

scala> implicit val files = df.inputFiles.flatMap(name => Map(name -> getModificationTime(name))).toMap
files: scala.collection.immutable.Map[String,Long] = Map(file:///tmp/par/part-00000-c6360540-c56d-48c4-8795-05a9c0ac4d18-c000_2.snappy.parquet -> 1588080295000, file:///tmp/par/part-00000-c6360540-c56d-48c4-8795-05a9c0ac4d18-c000_3.snappy.parquet -> 1588080299000, file:///tmp/par/part-00000-c6360540-c56d-48c4-8795-05a9c0ac4d18-c000_4.snappy.parquet -> 1588080302000, file:///tmp/par/part-00000-c6360540-c56d-48c4-8795-05a9c0ac4d18-c000.snappy.parquet -> 1588071322000)

scala> :paste
// Entering paste mode (ctrl-D to finish)

def getTime(fileName:String)(implicit files: Map[String,Long]): Long = {
 files.getOrElse(fileName,0L)
}

// Exiting paste mode, now interpreting.

getTime: (fileName: String)(implicit files: Map[String,Long])Long

scala> val input_file_modification_time = udf(getTime _)
input_file_modification_time: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,LongType,Some(List(StringType)))

scala> df.withColumn("createdDate",input_file_modification_time(input_file_name)).show
+---+-------------+
| id|  createdDate|
+---+-------------+
|  1|1588080295000|
|  2|1588080295000|
|  3|1588080295000|
|  4|1588080295000|
|  5|1588080295000|
|  6|1588080295000|
|  7|1588080295000|
|  8|1588080295000|
|  9|1588080295000|
| 10|1588080295000|
| 11|1588080295000|
| 12|1588080295000|
| 13|1588080295000|
| 14|1588080295000|
| 15|1588080295000|
| 16|1588080295000|
| 17|1588080295000|
| 18|1588080295000|
| 19|1588080295000|
| 20|1588080295000|
+---+-------------+
only showing top 20 rows


scala>

Answer 2

问题是，spark在UDF中为空，因为它只存在于驱动程序上。另一个问题是HadoopConfiguration是不可序列化的，因此不能轻易地将其封装到udf中。但是有一个使用org的训练。阿帕奇。火花SerializableWritable：

import org.apache.spark.SerializableWritable
import org.apache.hadoop.conf.Configuration

val conf = new SerializableWritable(spark.sparkContext.hadoopConfiguration)

def getModificationTime(path: String, conf:SerializableWritable[Configuration]): Long = {
    org.apache.hadoop.fs.FileSystem.get(conf.value)
        .getFileStatus(new org.apache.hadoop.fs.Path(path))
        .getModificationTime()
}

def input_file_modification_time(conf:SerializableWritable[Configuration]) = udf((path: String) => getModificationTime(path,conf))

data.select(input_file_modification_time(conf)($"input_file_name") as "modification_time").show(20, false)