Example: Generate new segments from silver data to the gold layer

Important

Some or all of this functionality is available as part of a preview release. The content and the functionality are subject to change.

This example shows a general pattern for generating a new gold segment from existing silver and gold data.

The scenario in this article creates an Engagement Recency segment that groups constituents by how recently they engaged.

Example segment values

This example uses these segment labels:

  • Engaged in Last 30 Days
  • Engaged in 31-90 Days
  • Engaged in 91-365 Days
  • Inactive
  • Unclassified

When to use this pattern

Use this pattern when you want to classify constituents into reusable analytical groups rather than storing a single derived metric.

Add the gold tables

Open the Fundraising_GD_CreateSchema notebook and add schema methods for the segment dimension and bridge table.

@staticmethod
def get_dimconstituentsegment_engagementrecency_schema() -> StructType:
    return StructType([
        StructField("ConstituentSegmentId", StringType()),
        StructField("ConstituentSegmentKey", LongType(), nullable=False),
        StructField("ConstituentSegmentName", StringType()),
        StructField("Order", IntegerType())
    ])

@staticmethod
def get_dimconstituentsegmentbridge_engagementrecency_schema() -> StructType:
    return StructType([
        StructField("ConstituentKey", LongType(), nullable=False),
        StructField("ConstituentSegmentKey", LongType(), nullable=False),
        StructField("ConstituentSegmentMappingId", StringType()),
        StructField("ConstituentSegmentBridgeKey", LongType(), nullable=False)
    ])

Register both tables in get_entities().

def get_entities(self) -> list[tuple[StructType, str, bool]]:
    return [
        # ... existing tables ...
        (NonprofitGoldModel.get_dimconstituentsegment_engagementrecency_schema(), "DimConstituentSegment_EngagementRecency", True),
        (NonprofitGoldModel.get_dimconstituentsegmentbridge_engagementrecency_schema(), "DimConstituentSegmentBridge_EngagementRecency", True),
    ]

Create the segment definitions

Open the Fundraising_GD_CreateSegments notebook and create the reusable segment definition rows.

from pyspark.sql import Row
from pyspark.sql.functions import col, xxhash64
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

segment_rows = [
    Row(ConstituentSegmentId="eng-recency-30", ConstituentSegmentName="Engaged in Last 30 Days", Order=1, DaysMin=0, DaysMax=30),
    Row(ConstituentSegmentId="eng-recency-31-90", ConstituentSegmentName="Engaged in 31-90 Days", Order=2, DaysMin=31, DaysMax=90),
    Row(ConstituentSegmentId="eng-recency-91-365", ConstituentSegmentName="Engaged in 91-365 Days", Order=3, DaysMin=91, DaysMax=365),
    Row(ConstituentSegmentId="eng-recency-inactive", ConstituentSegmentName="Inactive", Order=4, DaysMin=366, DaysMax=999999),
    Row(ConstituentSegmentId="eng-recency-unclassified", ConstituentSegmentName="Unclassified", Order=5, DaysMin=None, DaysMax=None),
]

segment_definition_schema = StructType([
    StructField("ConstituentSegmentId", StringType(), nullable=False),
    StructField("ConstituentSegmentName", StringType(), nullable=False),
    StructField("Order", IntegerType(), nullable=False),
    StructField("DaysMin", IntegerType()),
    StructField("DaysMax", IntegerType())
])

segment_definition_df = (
    spark.createDataFrame(segment_rows, schema=segment_definition_schema)
    .withColumn("ConstituentSegmentKey", xxhash64(col("ConstituentSegmentId")).cast("bigint"))
)

segment_definition_df.select(
    "ConstituentSegmentId",
    "ConstituentSegmentKey",
    "ConstituentSegmentName",
    "Order"
).write.mode("overwrite").format("delta").saveAsTable(
    f"{gold_lakehouse_name}.DimConstituentSegment_EngagementRecency"
)

The definition table stays reusable, while the DaysMin and DaysMax helper columns stay inside the notebook logic.

Build segment membership

Use the engagement timeline to calculate the latest engagement for each constituent and then map each constituent to one segment.

from delta.tables import DeltaTable
from pyspark.sql.functions import current_date, datediff, lit, max as spark_max, when, xxhash64

segment_df = segment_definition_df.select(
    "ConstituentSegmentKey",
    "ConstituentSegmentName",
    "DaysMin",
    "DaysMax"
)

latest_engagement_df = (
    get_gold_table("dm_EngagementTimeline")
    .join(
        get_gold_table("DimDate").select("DateKey", "Date"),
        col("EngagementDate") == col("DateKey"),
        "left"
    )
    .groupBy("ConstituentKey")
    .agg(spark_max("Date").alias("LastEngagementDate"))
    .withColumn("DaysSinceLastEngagement", datediff(current_date(), col("LastEngagementDate")))
)

classified_df = latest_engagement_df.join(
    segment_df,
    (col("DaysSinceLastEngagement") >= col("DaysMin")) &
    (col("DaysSinceLastEngagement") <= col("DaysMax")),
    "left"
)

unclassified_key = (
    segment_definition_df
    .filter(col("ConstituentSegmentName") == "Unclassified")
    .select("ConstituentSegmentKey")
    .first()["ConstituentSegmentKey"]
)

final_bridge_df = (
    classified_df
    .withColumn(
        "FinalSegmentKey",
        when(col("ConstituentSegmentKey").isNotNull(), col("ConstituentSegmentKey"))
        .otherwise(lit(unclassified_key))
    )
    .select(
        "ConstituentKey",
        col("FinalSegmentKey").alias("ConstituentSegmentKey")
    )
    .withColumn("ConstituentSegmentMappingId", lit(None).cast("string"))
    .withColumn(
        "ConstituentSegmentBridgeKey",
        xxhash64(col("ConstituentKey"), col("ConstituentSegmentKey")).cast("bigint")
    )
)

bridge_table = DeltaTable.forName(spark, f"{gold_lakehouse_name}.DimConstituentSegmentBridge_EngagementRecency")
bridge_table.delete("ConstituentSegmentMappingId IS NULL")

final_bridge_df.write.format("delta").mode("append").saveAsTable(
    f"{gold_lakehouse_name}.DimConstituentSegmentBridge_EngagementRecency"
)

This example depends on dm_EngagementTimeline, so run it after the gold enrichment logic that creates that data mart.

Verify the result

  1. Open DimConstituentSegment_EngagementRecency and confirm that all segment definitions exist.
  2. Open DimConstituentSegmentBridge_EngagementRecency and confirm that each ConstituentKey appears at most once.
  3. Confirm that recently engaged constituents land in Engaged in Last 30 Days.
  4. Confirm that inactive constituents land in Inactive or Unclassified.
  5. Confirm that rerunning the notebook replaces the custom bridge rows instead of duplicating them.

Considerations

  • Segment dimensions define the allowed segment values.
  • Bridge tables store constituent membership.
  • Prefer deterministic classification logic so reruns produce stable results.
  • If a constituent can belong to multiple segments, use a multi-row bridge pattern instead of forcing one final segment.
  • If the segment logic depends on a shared gold data mart such as dm_EngagementTimeline, document that notebook dependency explicitly.

See also

  • Last updated on