读取与分区名称冲突的列的火花数据

Question

我们在 s3 上使用分区读取以下文件路径

prefix/company=abcd/service=xyz/date=2021-01-01/file_01.json
prefix/company=abcd/service=xyz/date=2021-01-01/file_02.json
prefix/company=abcd/service=xyz/date=2021-01-01/file_03.json

当我用 pyspark 阅读这些内容时

self.spark \
    .read \
    .option("basePath", 'prefix') \
    .schema(self.schema) \
    .json(['company=abcd/service=xyz/date=2021-01-01/'])

所有文件都具有相同的架构，并作为行加载到表中。文件可能是这样的：

{"id": "foo", "color": "blue", "date": "2021-12-12"}

问题是有时文件的日期字段与我的分区代码冲突，例如date。所以我想知道是否可以在没有分区列的情况下加载文件，重命名 JSON 日期列，然后添加分区列。

决赛桌是：

| id  | color | file_date  | company | service | date       |
-------------------------------------------------------------
| foo | blue  | 2021-12-12 | abcd    | xyz     | 2021-01-01 |
| bar | red   | 2021-10-10 | abcd    | xyz     | 2021-01-01 |
| baz | green | 2021-08-08 | abcd    | xyz     | 2021-01-01 |

编辑：

更多信息：我有时有 5 或 6 个分区，日期是其中之一（不是最后一个）。而且我也需要一次读取多个日期分区。我传递给 Spark 的架构还包含分区列，这使得它更加复杂。

我不控制输入数据，所以我需要按原样阅读。我可以重命名文件列，但不能重命名分区列。

是否可以像加入 2 个数据帧时那样为文件列添加别名？

火花 3.1

Answer 1

一种方法是使用例如Hadoop FS API 列出prefix S3 路径下的文件，然后将该列表传递给spark.read。这样 Spark 不会将它们检测为分区，并且您可以在需要时重命名文件列。

将文件加载到数据框中后，循环遍历 df 列并重命名也存在于 partitions_colums 列表中的那些（例如，通过添加 file 前缀）。

最后，使用regexp_extract 函数解析来自input_file_name() 的分区。

这是一个例子：

from pyspark.sql import functions as F

Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
conf = sc._jsc.hadoopConfiguration()

s3_path = "s3://bucket/prefix"
file_cols = ["id", "color", "date"]
partitions_cols = ["company", "service", "date"]

# listing all files for input path
json_files = []
files = Path(s3_path).getFileSystem(conf).listFiles(Path(s3_path), True)

while files.hasNext():
    path = files.next().getPath()
    if path.getName().endswith(".json"):
        json_files.append(path.toString())

df = spark.read.json(json_files) # you can pass here the schema of the files without the partition columns

# renaming file column in if exists in partitions
df = df.select(*[
    F.col(c).alias(c) if c not in partitions_cols else F.col(c).alias(f"file_{c}")
    for c in df.columns
])

# parse partitions from filenames
for p in partitions_cols:
    df = df.withColumn(p, F.regexp_extract(F.input_file_name(), f"/{p}=([^/]+)/", 1))

df.show()

#+-----+----------+---+-------+-------+----------+
#|color| file_date| id|company|service|      date|
#+-----+----------+---+-------+-------+----------+
#|green|2021-08-08|baz|   abcd|    xyz|2021-01-01|
#| blue|2021-12-12|foo|   abcd|    xyz|2021-01-01|
#|  red|2021-10-10|bar|   abcd|    xyz|2021-01-01|
#+-----+----------+---+-------+-------+----------+

Answer 2

最简单的方法是更改​​分区列名。然后，您可以读取数据并根据需要重命名列。您也不会失去分区的好处。

如果这不是您可以在 jsons 中使用分区通配符读取的选项，请将日期列重命名为“file_date”，然后通过从文件名中提取分区日期来添加分区日期。您可以在pyspark.sql.functions 中从input_file_name 获取文件名。

编辑：我错过了您在日期之前还有其他分区列，您还必须从文件名中提取它们，使其不太理想。

Answer 3

是的，我们可以读取所有没有分区列的 json 文件。 直接使用父文件夹路径，它会将所有分区数据加载到数据框中。

读取数据框后，可以使用 withColumn() 函数重命名日期字段。

类似下面的东西应该可以工作

df= spark.read.json("s3://bucket/table/**/*.json")

renamedDF= df.withColumnRenamed("old column name","new column name")

Answer 4

在您的具体情况下，您的机会是读取 json 文件。

因此，不是加载没有分区列的文件，而是重命名 JSON 日期列，然后添加分区列，您可以加载没有文件列的分区列，重命名分区列，然后对文件数据应用架构。

您必须执行以下步骤：

• 使用sparkSession.read.text()将您的json文件作为文本读取
• 使用withColumnRenamed()更改分区列的名称
• 使用from_json方法将读取的json字符串转换为json
• 使用withColumnRenamed()根据需要重命名列

完整的代码如下：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DateType

# schema of json files
schema = StructType([
    StructField('id', StringType(), True),
    StructField('color', StringType(), True),
    StructField('date', DateType(), True)
])

df = sparkSession.read.text('resources') \
    .withColumnRenamed('date', 'partition_date') \
    .withColumn('json', F.from_json(F.col('value'), schema)) \
    .select('company', 'service', 'partition_date', 'json.*') \
    .withColumnRenamed('date', 'file_date') \
    .withColumnRenamed('partition_date', 'date')

示例

在prefix/company=abcd/service=xyz/date=2021-01-01目录下输入文件file_01.json和file_02.json

file_01.json

{"id": "foo", "color": "blue", "date": "2021-12-12"}
{"id": "bar", "color": "red", "date": "2021-12-13"}
{"id": "kix", "color": "yellow", "date": "2021-12-14"}

file_02.json

{"id": "kaz", "color": "blue", "date": "2021-12-15"}
{"id": "dir", "color": "red", "date": "2021-12-16"}
{"id": "tux", "color": "yellow", "date": "2021-12-17"}

你得到以下输出df数据框：

+-------+-------+----------+---+------+----------+
|company|service|      date| id| color| file_date|
+-------+-------+----------+---+------+----------+
|   abcd|    xyz|2021-01-01|kaz|  blue|2021-12-15|
|   abcd|    xyz|2021-01-01|dir|   red|2021-12-16|
|   abcd|    xyz|2021-01-01|tux|yellow|2021-12-17|
|   abcd|    xyz|2021-01-01|foo|  blue|2021-12-12|
|   abcd|    xyz|2021-01-01|bar|   red|2021-12-13|
|   abcd|    xyz|2021-01-01|kix|yellow|2021-12-14|
+-------+-------+----------+---+------+----------+