Hi team,
My team is struggling on “upsert” functio...
# troubleshootingy
yelim yu
10/21/2021, 3:15 PMHi team,
My team is struggling on “upsert” function.
Here is a problem.
We want to make realtime streaming user_state_table as an unique table while there are 3 different events which can update user_state_table. These 3 events have different columns but surely primary key and time dimension (create_time) are all included in these 3. The schema of user_state_table has intersecting columns of these 3 events.
Lets say event_1 has columns of (id, create_time, a,b) and event_2 has columns of (id, create_time, c,d). And we set the upsert mode as full. If we publish event_1 -> event_2 ->event_1 in that order, column c and d become null even though we want last event_1 ‘s column a&b overwrite first event_1 while event_2’s colunms are not being null.
How could we solve this problem by modifying table json & schema json? Please refer the comment to see how my team suggested schema.
Many thks from korea
yelim yu
10/21/2021, 3:32 PMuser state table
{
"tableName": "user_state",
"tableType": "REALTIME",
"segmentsConfig": {
"timeColumnName": "create_time",
"timeType": "SECONDS",
"segmentPushType": "APPEND",
"segmentAssignmentStrategy": "BalanceNumSegmentAssignmentStrategy",
"schemaName": "user_state",
"replicasPerPartition": "1"
},
"tenants": {},
"tableIndexConfig": {
"loadMode": "MMAP",
"streamConfigs": {
"streamType": "kafka",
"stream.kafka.consumer.type": "lowLevel",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.topic.name": "user.event",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
"stream.kafka.hlc.zk.connect.string": "z-1.kafka-engine-v2.robgzv.c2.kafka.ap-northeast-2.amazonaws.com:2181/kafka",
"stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
"stream.kafka.zk.broker.url": "z-3.kafka-engine-v2.robgzv.c2.kafka.ap-northeast-2.amazonaws.com:2181/kafka",
"stream.kafka.broker.list": "b-1.kafka-engine-v2.robgzv.c2.kafka.ap-northeast-2.amazonaws.com:9092,b-2.kafka-engine-v2.robgzv.c2.kafka.ap-northeast-2.amazonaws.com:9092,b-3.kafka-engine-v2.robgzv.c2.kafka.ap-northeast-2.amazonaws.com:9092"
},
"nullHandlingEnabled": true
},
"fieldConfigList": [],
"metadata": {
"customConfigs": {}
},
"routing": {
"instanceSelectorType": "strictReplicaGroup"
},
"upsertConfig": {
"mode":”FULL”
}
}
yelim yu
10/21/2021, 3:34 PMuser state schema
{
"schemaName": "user_state",
"primaryKeyColumns": [
"id"
],
"dimensionFieldSpecs": [
{
"name": "id",
"dataType": "STRING"
},
{
"name": "a",
"dataType": "STRING"
},
{
"name": "b",
"dataType": "STRING"
},
{
"name": "c",
"dataType": "STRING"
},
{
"name": "d",
"dataType": "STRING"
}
],
"metricFieldSpecs": [],
"dateTimeFieldSpecs": [
{
"name": "create_time",
"dataType": "STRING",
"format": "1DAYSSIMPLE_DATE_FORMATyyyy MM ddTHHmm:ss",
"granularity": "1:SECONDS"
}
]
}
k
Kenny Bastani
10/21/2021, 5:48 PMI'm not aware of any way to resolve this situation, since you have multiple different event schemas that you want to merge into a partial upsert. This does sound like an interesting use case though. Maybe it's something we can add to Pinot in the future.
Kenny Bastani
10/21/2021, 5:51 PMWe might be able to add a UDF called
FoldKenny Bastani
10/21/2021, 6:04 PM@Jackie What do you think?
j
Jackie
10/21/2021, 7:18 PMI think you should be able to use partial upsert and define all 4 columns as overwrite
Jackie
10/21/2021, 7:18 PMAdd @Yupeng Fu
Jackie
10/21/2021, 7:19 PMk
Kenny Bastani
10/21/2021, 7:26 PMThanks @Jackie. I haven’t played with that yet. Seems like it should work as described for this scenario.
Kenny Bastani
10/21/2021, 7:30 PMIt does look like the result will be returned as a set which will still include the new item. I will test this out later.
y
yelim yu
10/21/2021, 11:28 PMMy team will get start to do it! Let me share the result!!
👍 3
k
Kenny Bastani
10/22/2021, 12:38 AMGreat @yelim yu. 🙂