从Databricks Autoloader获取已加载文件的列表

Question

我们可以使用Autoloader来跟踪是否已从S3存储桶加载的文件。我关于Autoloader的问题：有没有办法读取Autoloader数据库以获取已加载文件的列表？

我可以在AWS Glue job书签中轻松做到这一点，但我不知道如何在Databricks Autoloader中做到这一点。

Answer 1

如果使用checkpointLocation选项，则可以通过读取rocksDB日志来读取所有已处理的文件。一些示例代码实现了这一点，请注意，您需要指向检查点位置上的路径，以便检索加载的文件列表。

from glob import glob
import codecs

directory = "<YOUR_PATH_GOES_HERE>/sources/*/rocksdb/logs/"
for file in glob(f"{directory}/*.log"):
    with codecs.open(file, encoding='utf-8', errors='ignore') as f:
        f = f.readlines()
        print(f)

PS。:需要正确解析日志，以便只获得文件名。

Answer 2

.load("path")
.withColumn("filePath",input_file_name())

例如，您可以将filePath插入到您的流接收器中，然后从那里获取不同的值，或者使用forEatch / forEatchBatch，并将其插入到spark sql表中