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

Question

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

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

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

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

Answer 1

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

我们这里的策略将归结为以下几点：

对于每个输入，构造一个指示每个唯一行的索引（行号）
融合数据集，记下每行的原始列名和单元格值
将所有融化的数据集堆叠成一个长数据集
将长数据集中的列名（现在，这些是行）更改为所需的别名
将长数据集转换为最终的协调数据集

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

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

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

我们创建示例数据帧的设置如下所示：

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 行应该在输出中显示为两个单独的行。为此，我们跟踪一行的 origin

构建索引

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

注意：不要使用 'naked window'（无分区列）即 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|
+-----+-------------+-----+------+-----+------+
"""

熔化数据帧

我们可以使用技术创建另一个数据集。方便的是，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 调用放在彼此之上，然后是最终的 'otherwise' ，我们就地使用列名。这让我们可以高效地运行遍历每一行并确定它包含的列是否是需要重命名的别名。

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|
+-----+-----+------+-----+
"""

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

How do I union many distinct schemas into a single output I can dynamically pivot later?

palantir-foundry

foundry-code-repositories

foundry-python-transform

构建索引

熔化数据帧

联合数据帧

别名列

转向输出