如何将Dagster分区资产合并为单一DataFrame？

我有一个软件定义资产，用于将大量分区数据与单个小型DataFrame进行对比，示例代码如下：

partitions = dagster.StaticPartitionsDefinition(["a", "b"])

@dagster.asset(partitions_def=partitions)
def large_dataframes(context):
    return pd.DataFrame(np.random.randint(0,100,size=(100, 4)), columns=list('ABCD'))

@dagster.asset
def small_dataframe(context):
    return pd.DataFrame(np.random.randint(0,100,size=(100, 4)), columns=list('DEFG'))

@dagster.asset(partitions_def=partitions)
def one_df_filtered_by_the_other(context, large_dataframes, small_dataframe):
    return large_dataframes.join(small_dataframe, rsuffix=".")

代码运行正常，但第三个资产的输出为分区DataFrame，实际中多数分区为空，非空分区数据量极小。我希望将这些分区合并为单一DataFrame，作为可被下游依赖的新资产。

我当前的实现方式如下：

@dagster.asset
def reduce_to_unpartitioned(context,one_df_filtered_by_the_other):
    return pd.concat(one_df_filtered_by_the_other.values())

但这个方法感觉像是临时方案，且担心在生产环境（分区数量达数百或数千个）中是否能正常工作，因为未在文档中找到官方方法。请问此实现是否正确？是否有更优方案？

当前实现的正确性 #

你的实现完全正确。当非分区资产依赖分区资产时，Dagster会自动将所有分区的结果以字典形式传递给资产函数（键为分区键，值为对应分区的数据），通过pd.concat合并字典中的DataFrame完全符合Dagster的设计逻辑。

生产环境适用性 #

针对数百甚至数千个分区的场景，该方案的可行性取决于两个核心因素：

• 内存限制：若所有非空分区的总数据量在内存可承受范围内，pd.concat可以正常工作；若数据量过大，建议合并前过滤空分区，减少不必要的内存占用：

  @dagster.asset
  def reduce_to_unpartitioned(context, one_df_filtered_by_the_other):
      # 过滤空DataFrame，仅保留有数据的分区
      non_empty_dfs = [df for df in one_df_filtered_by_the_other.values() if not df.empty]
      return pd.concat(non_empty_dfs) if non_empty_dfs else pd.DataFrame()
  

• 性能优化：如果分区数量极多，可结合DynamicPartitionsDefinition实现批量处理，但对于你这种多数分区为空的场景，提前过滤空分区已能大幅提升合并效率。

更优实践方案 #

如果希望更贴合Dagster的最佳实践，可考虑以下两种优化方向：

1. 在分区资产阶段跳过空结果存储：
   在one_df_filtered_by_the_other资产中，当结果为空时调用context.skip()跳过存储，这样后续合并时字典中只会包含非空分区的数据，既减少存储占用，也提升合并效率：

   @dagster.asset(partitions_def=partitions)
   def one_df_filtered_by_the_other(context, large_dataframes, small_dataframe):
       result = large_dataframes.join(small_dataframe, rsuffix=".")
       if result.empty:
           context.skip()
       return result
   

2. 按需加载分区数据：
   若无需合并所有分区，可通过Dagster的资产选择API手动指定需要合并的分区，但对于常规的全量合并场景，直接依赖分区资产并合并的方式已经足够简洁高效。

内容的提问来源于stack exchange，提问作者Mike Dewar