如何在镶木地板分区中有不同的模式

Question

我有 json 个文件读入数据框。 json 可以有一个特定于名称的结构字段消息，如下所示。

Json1
{
   "ts":"2020-05-17T00:00:03Z",
   "name":"foo",
   "messages":[
      {
         "a":1810,
         "b":"hello",
         "c":390
      }
   ]
}

Json2
{
   "ts":"2020-05-17T00:00:03Z",
   "name":"bar",
   "messages":[
      {
         "b":"my",
         "d":"world"
      }
   ]
}

当我将 json 中的数据读入 Dataframe 时，我得到如下所示的架构。

root
 |-- ts: string (nullable = true)
 |-- name: string (nullable = true)
 |-- messages: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: long (nullable = true)
 |    |    |-- b: string (nullable = true)
 |    |    |-- c: long (nullable = true)
 |    |    |-- d: string (nullable = true)

这很好。现在，当我保存到按名称分区的 parquet 文件时，如何在 foo 和 bar 分区中有不同的模式？

path/name=foo
root
 |-- ts: string (nullable = true)
 |-- name: string (nullable = true)
 |-- messages: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: long (nullable = true)
 |    |    |-- b: string (nullable = true)
 |    |    |-- c: long (nullable = true)

path/name=bar
root
 |-- ts: string (nullable = true)
 |-- name: string (nullable = true)
 |-- messages: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- b: string (nullable = true)
 |    |    |-- d: string (nullable = true)

当我从根路径读取数据时，如果我得到包含 foo 和 bar 的所有字段的架构，我很好。但是当我从 path/name=foo 读取数据时，我只期待 foo 模式。

Answer 1

1. Partitioning & Storing as Parquet file:

如果您保存为 parquet 格式，那么在读取 path/name=foo 包括 all 必填字段(a,b,c)，然后spark只加载那些字段。

如果我们 won't 指定架构，那么 所有字段(a,b,c,d) 都将是包含在数据框中

EX:

schema=define structtype...schema
spark.read.schema(schema).parquet(path/name=foo).printSchema()

2.Partitioning & Storing as JSON/CSV file:

然后 Spark 不会将 b,d 列 添加到 path/name=foo 文件中，所以当我们只读取 name=foo 目录，我们将不会在数据中包含 b,d 列。

EX:

spark.read.json(path/name=foo).printSchema()
spark.read.csv(path/name=foo).printSchema()

Answer 2

您可以在将数据框保存到分区之前更改架构，为此您必须过滤分区记录，然后将它们保存到相应的文件夹中

#this will select only not null columns which will drop col d from foo and a,c from bar
df = df.filter(f.col('name')='foo').select(*[c for c in df.columns if df.filter(f.col(c).isNotNull()).count() > 0])

#then save the df
df.write.json('path/name=foo')

现在每个分区都会有不同的架构。

如何在镶木地板分区中有不同的模式

How to have different schemas within parquet partitions

apache-spark

parquet

apache-spark-sql