Update metadata export logic for join derivation #879
Conversation
| tableUtils.createDatabase(namespace) | ||
| val viewsGroupBy = getViewsGroupBy(suffix = "cumulative", makeCumulative = true, namespace) | ||
| val joinConf = getEventsEventsTemporal("cumulative", namespace) | ||
| joinConf.setJoinParts(Seq(Builders.JoinPart(groupBy = viewsGroupBy)).asJava) |
There was a problem hiding this comment.
can you add a test case that contains external parts?
| StructType(derivedDummyOutputDf.schema.filterNot(f => keyAndPartitionFields.map(_.name).contains(f.name)))) | ||
| ListBuffer(columns.map { tup => toAggregationMetadata(tup._1, tup._2, joinConf.hasDerivations) }: _*) | ||
| } else { | ||
| aggregationsMetadata |
There was a problem hiding this comment.
nit: rename agg because agg is specific to group_by, but now we have external parts and derivations
aggMetadata=>joinOutputFieldsMetadataaggregationsMetadata=>joinIntermediateFieldsMetadata
There was a problem hiding this comment.
The concept of this aggMetadata is more like output features(not including keys). It will be used as the source data of features on MLI tool. I am thinking of maybe renaming this to featuresMetadata or joinOutputValuesMetadata.
There was a problem hiding this comment.
@yuli-han rename?
aggMetadata=>joinOutputValuesMetadataaggregationsMetadata=>joinIntermediateValuesMetadata
| expression = "*" | ||
| ), Derivation( | ||
| name = "test_feature_name", | ||
| expression = f"${viewsGroupBy.metaData.name}_time_spent_ms_average" |
There was a problem hiding this comment.
Can you add some derivations that use ts and ds as inputs?
There was a problem hiding this comment.
Similarly, can we add a test case for key columns as output?
such as
Derivation(
name = "event_id",
expression = "ext_contextual_event_id"
)
There was a problem hiding this comment.
Plus one on the ts and ds expressions. Those values will be derived from the request time.
| val finalOutputColumns = joinConf.derivationsScala.finalOutputColumn(dummyOutputDf.columns).toSeq | ||
| val derivedDummyOutputDf = dummyOutputDf.select(finalOutputColumns: _*) | ||
| val columns = SparkConversions.toChrononSchema( | ||
| StructType(derivedDummyOutputDf.schema.filterNot(f => keyAndPartitionFields.map(_.name).contains(f.name)))) |
There was a problem hiding this comment.
Is the filterNot necessary here? I think we should keep everything that users included in derivations. For example, we should allow key columns to be in the output if users explicitly included it in derivations.
There was a problem hiding this comment.
The aggMetadata are actually features, so the purpose of this line is to exclude the keys from the features list.
But this may not apply to the case where the users rename an external feature to a key name, if that happens the key(which is actually an external feature) will be filtered out here 🤔
| Map("ts" -> api.StringType, "ds" -> api.StringType) | ||
| } | ||
| val sparkSchema = { | ||
| val keySchema = leftSchema.filter(tup => keyColumns.contains(tup._1)) |
There was a problem hiding this comment.
we will need to handle key mapping here
There was a problem hiding this comment.
handled in above code
| val keyCols: Seq[String] = joinPart.groupBy.keyColumns.toScala | ||
| if (joinPart.keyMapping == null || joinPart.keyMapping.isEmpty) keyCols | ||
| else { | ||
| keyCols.map(key => joinPart.keyMapping.getOrDefault(key, key)) |
There was a problem hiding this comment.
@yuli-han keyMapping is a mapping from left_key to right_key, but here you are doing the reverse
There was a problem hiding this comment.
Let's include this in a unit test
There was a problem hiding this comment.
thanks for catching! Updated the code and also added a unit test to cover it. Also tested locally using a manually created testing config and attached the result in the above testing result doc.
| val findKey = joinPart.keyMapping.toScala.find(_._2 == key) | ||
| if (findKey.isDefined) { | ||
| findKey.get._1 | ||
| } else { | ||
| key | ||
| } |
There was a problem hiding this comment.
use joinPart.rightToLeft: https://github.com/airbnb/chronon/blob/main/api/src/main/scala/ai/chronon/api/Extensions.scala#L726
Summary
Update the metadata export logic for join with derivation. Use derived columns as exported features.
Why / Goal
Test Plan
Testing result using Airbnb internal config https://docs.google.com/document/d/1RYMioAdAZxAqEjyfgQBenFFGXprAlhUsfHhJaYA-L8Y/edit?tab=t.0
Checklist
Reviewers
@hzding621 @pengyu-hou @SophieYu41