palantir-foundry - 如何将许多不同的模式联合到一个输出中，以便稍后动态转换？

Question

我想采用任意一组模式并将它们组合成一个数据集，以后可以取消透视。最稳定的方法是什么？

假设我有数十个具有不同架构的输入。这些都有不同的列，实际上意味着相同的东西，它们只是没有相同的名称。

修复列名称后，我想创建作为这些列集合的输出，即动态旋转。我希望我的输入组合在一起后有一个单一的模式，我可以在以后做这个动态旋转。

我真的不想遍历并克隆输入或创建数百个中间数据集，那么我最好怎么做呢？

Answer 1

可用于协调任意数据集架构的一种策略是使用单元级技术，在该技术中，您制作一个长数据集，记录每个单元格的内容。

我们的策略归结为以下几点:

1. 对于每个输入，构建一个索引(行号)来指示每个唯一行
2. 融合数据集，记下每行的原始列名和单元格值
3. 将所有融化的数据集堆叠成一个长数据集
4. 将长数据集中的列名(现在是行)更改为所需的别名
5. 将长数据集转换为最终的协调数据集

注意:我们采用此策略而不是使用一堆 .withColumn 调用(我们直接重命名数据帧的列)的原因是允许在我们的管道下游进行更多动态的反旋转。

如果我们想象下游的消费者可能想要描述他们自己的别名规则或列组合，我们需要一个包含所有单元格、列和原始行号的长数据集以从中提取。

如果我们不需要动态旋转，采用每个输入数据集并重命名列的更简单策略就可以了。

我们创建示例 DataFrame 的设置如下所示:

from pyspark.sql import types as T, functions as F, window, SparkSession

spark = SparkSession.builder.getOrCreate()


# Synthesize DataFrames
schema1 = T.StructType([
  T.StructField("col_1_alias_1", T.StringType(), False),
  T.StructField("col_2", T.IntegerType(), False),
  T.StructField("col_3", T.StringType(), False),
  T.StructField("col_4", T.IntegerType(), False),
])
data1 = [
  {"col_1_alias_1": "key_1", "col_2": 1, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_1": "key_2", "col_2": 2, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_1": "key_3", "col_2": 3, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_1": "key_1", "col_2": 1, "col_3": "UPDATE", "col_4": 1},
  {"col_1_alias_1": "key_2", "col_2": 2, "col_3": "UPDATE", "col_4": 1},
  {"col_1_alias_1": "key_1", "col_2": 1, "col_3": "DELETE", "col_4": 2},
]

df1 = spark.createDataFrame(data1, schema1)

df1 = df1.withColumn("origin", F.lit("df1"))



schema2 = T.StructType([
  T.StructField("col_1_alias_2", T.StringType(), False),
  T.StructField("col_2", T.IntegerType(), False),
  T.StructField("col_3", T.StringType(), False),
  T.StructField("col_4", T.IntegerType(), False),
])
data2 = [
  {"col_1_alias_2": "key_1", "col_2": 1, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_2": "key_2", "col_2": 2, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_2": "key_3", "col_2": 3, "col_3": "CREATE", "col_4": 0},
  {"col_1_alias_2": "key_1", "col_2": 1, "col_3": "UPDATE", "col_4": 1},
  {"col_1_alias_2": "key_2", "col_2": 2, "col_3": "UPDATE", "col_4": 1},
  {"col_1_alias_2": "key_1", "col_2": 1, "col_3": "DELETE", "col_4": 2},
]

df2 = spark.createDataFrame(data2, schema2)

df2 = df2.withColumn("origin", F.lit("df2"))

注意:我们会跟踪 DataFrame 的来源，这样当我们逆透视时，我们不会不小心将同一行的两个不同来源的数据拉到同一行中。即 df1 的第 0 行和 df2 的第 0 行应该在输出中显示为两个单独的行。为此，我们跟踪一行的来源

构建索引

我们首先要跟踪数据集的行号，以便将相应的单元格放在一起。

注意:请勿使用“裸窗口”(无分区列)，即在此处调用 Window.partitionBy()。这样做将导致一个任务执行整个 DataFrame 的 row_number 计算。如果你的 DataFrame 很大，这个将导致 OOM 并且您的构建无法很好地扩展。我们在这里选择使用较低级别的函数 zipWithIndex，虽然不太优雅，但可扩展性更好。

def create_index(df):
    schema_in = df.columns
    indexed = df.rdd.zipWithIndex().toDF().select(
        F.col("_2").alias("index"),
        *[
            F.col("_1").getItem(x).alias(x) for x in schema_in
        ]
    )
    return indexed


indexed1 = create_index(df1)
indexed1.show()
"""
+-----+-------------+-----+------+-----+------+
|index|col_1_alias_1|col_2| col_3|col_4|origin|
+-----+-------------+-----+------+-----+------+
|    0|        key_1|    1|CREATE|    0|   df1|
|    1|        key_2|    2|CREATE|    0|   df1|
|    2|        key_3|    3|CREATE|    0|   df1|
|    3|        key_1|    1|UPDATE|    1|   df1|
|    4|        key_2|    2|UPDATE|    1|   df1|
|    5|        key_1|    1|DELETE|    2|   df1|
+-----+-------------+-----+------+-----+------+
"""

indexed2 = create_index(df2)
indexed2.show()
"""
+-----+-------------+-----+------+-----+------+
|index|col_1_alias_2|col_2| col_3|col_4|origin|
+-----+-------------+-----+------+-----+------+
|    0|        key_1|    1|CREATE|    0|   df2|
|    1|        key_2|    2|CREATE|    0|   df2|
|    2|        key_3|    3|CREATE|    0|   df2|
|    3|        key_1|    1|UPDATE|    1|   df2|
|    4|        key_2|    2|UPDATE|    1|   df2|
|    5|        key_1|    1|DELETE|    2|   df2|
+-----+-------------+-----+------+-----+------+
"""

熔化数据框

我们可以使用 melt 创建另一个数据集技术。方便的是，transforms.verbs 包(默认导入到您的存储库中)有一个实用方法 transforms.verbs.unpivot，它具有您可以使用的等效实现。

from transforms.verbs import unpivot

unpivoted1 = unpivot(indexed1, id_vars=["index", "origin"], value_vars=indexed1.columns)
unpivoted1.show()
"""
+-----+------+-------------+------+
|index|origin|     variable| value|
+-----+------+-------------+------+
|    0|   df1|        index|     0|
|    0|   df1|col_1_alias_1| key_1|
|    0|   df1|        col_2|     1|
|    0|   df1|        col_3|CREATE|
|    0|   df1|        col_4|     0|
|    0|   df1|       origin|   df1|
|    1|   df1|        index|     1|
|    1|   df1|col_1_alias_1| key_2|
|    1|   df1|        col_2|     2|
|    1|   df1|        col_3|CREATE|
|    1|   df1|        col_4|     0|
|    1|   df1|       origin|   df1|
|    2|   df1|        index|     2|
|    2|   df1|col_1_alias_1| key_3|
|    2|   df1|        col_2|     3|
|    2|   df1|        col_3|CREATE|
|    2|   df1|        col_4|     0|
|    2|   df1|       origin|   df1|
|    3|   df1|        index|     3|
|    3|   df1|col_1_alias_1| key_1|
+-----+------+-------------+------+
...
"""
unpivoted2 = unpivot(indexed2, id_vars=["index", "origin"], value_vars=indexed2.columns)
unpivoted2.show()
"""
+-----+------+-------------+------+
|index|origin|     variable| value|
+-----+------+-------------+------+
|    0|   df2|        index|     0|
|    0|   df2|col_1_alias_2| key_1|
|    0|   df2|        col_2|     1|
|    0|   df2|        col_3|CREATE|
|    0|   df2|        col_4|     0|
|    0|   df2|       origin|   df2|
|    1|   df2|        index|     1|
|    1|   df2|col_1_alias_2| key_2|
|    1|   df2|        col_2|     2|
|    1|   df2|        col_3|CREATE|
|    1|   df2|        col_4|     0|
|    1|   df2|       origin|   df2|
|    2|   df2|        index|     2|
|    2|   df2|col_1_alias_2| key_3|
|    2|   df2|        col_2|     3|
|    2|   df2|        col_3|CREATE|
|    2|   df2|        col_4|     0|
|    2|   df2|       origin|   df2|
|    3|   df2|        index|     3|
|    3|   df2|col_1_alias_2| key_1|
+-----+------+-------------+------+
...
"""

联合数据框

这部分很简单，只需将您的 DataFrame 堆叠在一起即可。如果您有很多，请使用 transforms.verbs.dataframes.union_many

all_dfs = unpivoted1.unionByName(unpivoted2)

别名列

下一部分有点密集。

我们想要更改注释列的行的值，这意味着相同的事情，所以我们想要构建一个 case 语句，用于用别名替换它们的最终名称。

为此，我们需要一个大的 case 语句，这是通过将 .when() 语句堆叠在一起来完成的。

我们可以选择在 for() 循环之外创建一个 None 变量并检测第一次迭代 + 直接赋值，但在这种情况下我们可以简单地使用 pyspark.sql.functions 直接作为第一个“when”语句导入。这让我们可以将 when 调用放在彼此之上，然后是最后的“否则”，我们就地使用列名。这让我们可以高效地遍历每一行并确定它包含的列是否是需要重命名的别名。

alias_dictionary = {
    "col_1": ["col_1_alias_1", "col_1_alias_2"]
}

when_statement = F

for alias_key, alias_list in alias_dictionary.items():
    # For each alias, if the row we are on is a cell that needs to be aliased,
    #   i.e. it isin() a list, then we rename it to the alias.  Otherwise we 
    #   leave it alone and will use the final .otherwise() outside the loop
    when_statement = when_statement.when(
        F.col("variable").isin(alias_list), alias_key
    )
when_statement = when_statement.otherwise(F.col("variable"))

# Replace the column names with their aliases
all_dfs = all_dfs.withColumn("variable", when_statement)
all_dfs.show()

"""
+-----+------+--------+------+
|index|origin|variable| value|
+-----+------+--------+------+
|    0|   df1|   index|     0|
|    0|   df1|   col_1| key_1|
|    0|   df1|   col_2|     1|
|    0|   df1|   col_3|CREATE|
|    0|   df1|   col_4|     0|
|    0|   df1|  origin|   df1|
|    1|   df1|   index|     1|
|    1|   df1|   col_1| key_2|
|    1|   df1|   col_2|     2|
|    1|   df1|   col_3|CREATE|
|    1|   df1|   col_4|     0|
|    1|   df1|  origin|   df1|
|    2|   df1|   index|     2|
|    2|   df1|   col_1| key_3|
|    2|   df1|   col_2|     3|
|    2|   df1|   col_3|CREATE|
|    2|   df1|   col_4|     0|
|    2|   df1|  origin|   df1|
|    3|   df1|   index|     3|
|    3|   df1|   col_1| key_1|
+-----+------+--------+------+
...
"""

转向输出

最后，我们按每一行分组，将行转回列，取找到的值，瞧!

pivoted = all_dfs.groupBy(  # for each original row...
    "index", "origin"
).pivot(                    # pivot the rows back to columns...
    "variable"              
).agg(                      # don't perform any calculations, just take the cell...
    F.first("value")
).orderBy(                  # for printing reasons, order by the original row number...
    "index", "origin"                 
).drop(                     # and remove the index
    "index", "origin"                 
)

pivoted.show()
"""
+-----+-----+------+-----+
|col_1|col_2| col_3|col_4|
+-----+-----+------+-----+
|key_1|    1|CREATE|    0|
|key_1|    1|CREATE|    0|
|key_2|    2|CREATE|    0|
|key_2|    2|CREATE|    0|
|key_3|    3|CREATE|    0|
|key_3|    3|CREATE|    0|
|key_1|    1|UPDATE|    1|
|key_1|    1|UPDATE|    1|
|key_2|    2|UPDATE|    1|
|key_2|    2|UPDATE|    1|
|key_1|    1|DELETE|    2|
|key_1|    1|DELETE|    2|
+-----+-----+------+-----+
"""