Hi, I am using Flink 1.15 and have a use case of DynamoDb sink. Flink 1.15 does have DynamoDb connector but in connector option it didn't say anything about

SortKey

. Will the connector provide any option to get the "latest" data for a composite key? Any pointers would be very helpful. Many Thanks.

I havent used it DynamoDbSink. If the options which you are looking for is not present. You can always create a process function or a rich flatmap function operator to have this capability created. Instantiate the connection object on open() and use it for any other operation in your process function with the incoming events

Will the connector provide any option to get the “latest” data for a composite key?

Hey, this sounds like you want to read from DynamoDb. If so, the DDB sink is not what you want. We can read from DynamoDbStreams However, if you are writing to DynamoDB, we use the BatchWriteRecords API so it doesn’t care about specifying sort/primary key https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html

Thanks for the replies. Actually we are using DynamoDb for sinking only. And our usecase is "conditional put" for streaming data. We are using Table API. We can use composite key(main primary key and sort key in the connector) which might work(though need testing) but we need to insert "only" those records whose timestamp is greater than("latest") the existing timestamp of that composite key. Similar to below :

aws dynamodb update-item \
    --table-name ProductCatalog \
    --key '{"Id": {"N": "456"}}' \
    --update-expression "SET Price = Price - :discount" \
    --condition-expression "Price > :limit" \
    --expression-attribute-values <file://values.json>

And we don't want to do "sort" inside flink api which might have performance implications. Any idea? Thanks.

Ah I see… yes conditional put is not supported for DDB’s batch write API. We considered supporting conditional writes in future versions but the performance will probably be impacted because the API is single record instead of batch