如何根据文件模式将不同的文件加载到不同的表中？

Question

我正在运行一个简单的 PySpark 脚本，就像这样。

base_path = '/mnt/rawdata/'
file_names = ['2018/01/01/ABC1_20180101.gz',
               '2018/01/02/ABC2_20180102.gz',
               '2018/01/03/ABC3_20180103.gz',
               '2018/01/01/XYZ1_20180101.gz'
               '2018/01/02/XYZ1_20180102.gz']

for f in file_names:
  print(f)

所以，只要测试一下，我就可以找到文件并打印字符串就好了。现在，我试图弄清楚如何将每个文件的内容加载到 SQL Server 中的特定表中。问题是，我想对匹配模式的文件进行通配符搜索，并将特定文件加载到特定表中。所以，我想做以下事情：

1. 将名称中包含“ABC”的所有文件加载到我的“ABC_Table”中，并将名称中包含“XYZ”的所有文件加载到我的“XYZ_Table”中（所有数据从第 2 行开始，而不是第 1 行）
2. 将文件名加载到每个相应表中名为“file_name”的字段中（我完全可以使用“file_names”中的整个字符串或最后一个“/”字符之后的字符串部分；没关系)

为此我尝试使用Azure数据工厂，它可以递归循环所有文件就好了，但它没有加载文件名，我真的需要表中的文件名来区分哪些记录是来自哪些文件和日期。是否可以使用 Azure Databricks 执行此操作？我觉得这是一个可实现的 ETL 流程，但我对 ADB 的了解还不够，无法完成这项工作。

根据 Daniel 的建议更新

dfCW = sc.sequenceFile('/mnt/rawdata/2018/01/01/ABC%.gz/').toDF()
dfCW.withColumn('input', input_file_name())
print(dfCW)

给我：

com.databricks.backend.daemon.data.common.InvalidMountException:

接下来我可以尝试什么？

Answer 1

您可以从pyspark.sql.functions 使用input_file_name 例如

withFiles = df.withColumn("file", input_file_name())

之后，您可以通过过滤新列来创建多个数据框

abc = withFiles.filter(col("file").like("%ABC%"))
xyz = withFiles.filter(col("file").like("%XYZ%"))

然后对它们都使用常规编写器。