在合并语句数据块中的情况答案

【问题标题】：case when in merge statement databricks在合并语句数据块中的情况
【发布时间】：2022-12-22 19:30:30
【问题描述】：

我正在尝试使用 pyspark 中的合并语句在 Databricks 中进行更新。我想知道在 whenMatchedUpdate 部分是否允许使用表达式（例如添加两列，case when）。例如我想做这样的事情

deltaTableTarget = DeltaTable.forPath(spark, delta_table_path)

deltaTableTarget.alias('TgtCrmUserAggr') \
  .merge(
    broadcast(df_transformed.alias('DeltaSource')),
    "DeltaSource.primary_key==TargetTable.primary_key"
  ) \
   .whenMatchedUpdate(set =
    {
      "aggcount":"DeltaSource.count + TargetTable.count",
      "max_date": "case when DeltaSource.max_date > TargetTable.max_date then DeltaSource.max_date else TargetTable.max_date end"
    }
  ) 
  .whenNotMatchedInsert().insertAll()
)\
 .execute()

【问题讨论】：

标签： pyspark databricks

【解决方案1】：

如果我很好地理解你的逻辑，你可以只取 2 列的最大值，对吗？

deltaTableTarget = DeltaTable.forPath(spark, delta_table_path)

deltaTableTarget.alias('TgtCrmUserAggr') 
  .merge(
    broadcast(df_transformed.alias('DeltaSource')),
    "DeltaSource.primary_key==TargetTable.primary_key"
  ) 
   .whenMatchedUpdate(set =
    {
      "aggcount":"DeltaSource.count + TargetTable.count",
      "max_date": "MAX(DeltaSource.max_date,TargetTable.max_date)"
    }
  ) 
  .whenNotMatchedInsert().insertAll()
)
 .execute()

如果这不正确，您可以使用多个带有条件的 whenMatchedUpdate() 函数。

deltaTableTarget = DeltaTable.forPath(spark, delta_table_path)

deltaTableTarget.alias('TgtCrmUserAggr') 
  .merge(
    broadcast(df_transformed.alias('DeltaSource')),
    "DeltaSource.primary_key==TargetTable.primary_key"
  ) 
  .whenMatchedUpdate(condition= 'DeltaSource.max_date > TargetTable.max_date', 
     set =
         {
           "aggcount":"DeltaSource.count + TargetTable.count",
           "max_date": "DeltaSource.max_date"
          }
  ) 
.whenMatchedUpdate(set =
    {
      "aggcount":"DeltaSource.count + TargetTable.count",
      "max_date": "TargetTable.max_date"
     }
  ) 
  .whenNotMatchedInsert().insertAll()
)
 .execute()

【讨论】：