HI Laurent,
Did you manage to find the error in your MATCH_RECOGNIZE statement? If I
had to take a guess, I'd say it's because you are accessing A, but due to
the quantifier of * there might actually be no event A.
Cheers,
Konstantin
On Fri, Nov 27, 2020 at 10:03 PM Laurent Exsteens <
laurent
Hi Leonard,
> From my understanding, your case is not a pure deduplication case but
> want to both keep the previous record and current record, thus the
> deduplication query can not satisfy your requirement.
>
Indeed, that's what I came to realise during our discussion on this email
chain. I'm
Hi Laurent,
With respect to Ververica Platform, we will support Flink 1.12 and add
"upsert-kafka" as a packaged connector in our next minor release which we
target for February.
Cheers,
Konstantin
On Thu, Nov 12, 2020 at 3:43 AM Jark Wu wrote:
> Hi Laurent,
>
> 1. Deduplicate with keeping the
Hi, Laurent
>
> I'm not sure that would do what I want though. As far as I understand, the
> deduplication query will always remember any values it has seen. So if I
> have, for a specific primary key, the following values in another field: "a",
> "a", "b", "b", "a", "a", the deduplication quer
Hi Leonard,
thank you for your answer.
I'm not sure that would do what I want though. As far as I understand, the
deduplication query will always remember any values it has seen. So if I
have, for a specific primary key, the following values in another field:
"a", "a", "b", "b", "a", "a", the ded
Hi, Laurent
> Basically, I need to deduplicate, but only keeping in the deduplication state
> the latest value of the changed column to compare with. While here it seems
> to keep all previous values…
You can use ` ORDER BY proctime() DESC` in the deduplication query, it will
keep last row,
Hello,
seems like LAG would probably be the right function to use.
However, I get unexpected results from it:
---
-- Deduplicate addresses
---
--CREATE TEMPORARY VIEW dedup_address as
SELECT *
FROM (
SELECT client_number
, address
,
I'm now trying with a MATCH_RECOGNIZE:
SELECT *
FROM customers
MATCH_RECOGNIZE (
PARTITION BY client_number
ORDER BY proctime()
MEASURES
LAST(B.client_number) as client_number,
LAST(B.address) as address
PATTERN (A* B)
DEFINE
B AS LAST(A.address, 1) is NULL OR B.address <> LAST(A.address, 1)
) as
I see what was my mistake: I was using a field in my ORDER BY, while it
only support proctime() for now.
That allows me to create an append only stream, thanks a lot!
However, it still does not allow me to do what I need:
*If I use both my primary key and changing column in PARTITION BY, then i
Hi Laurent,
1. Currently, it's impossible to convert deduplicate with last row into an
append-only stream.
2. Yes, I think Ververica platform doesn't support 'changelog-json' format
natively.
However, regarding your case, I think you can use keep first row on
client_number+address key.
SELECT *
Hi Jark,
thanks again for your quick response!
I tried multiple variants of my query by:
- specifying only the primary key in the PARTITION BY clause
- changing the order to DESC to keep the last row
--> I unfortunately always get the same error message.
If I try to make a simple select on the r
Hi Laurent,
1. Deduplicate with keeping the first row will generate an append-only
stream. But I guess you are expecting to keep the last row which generates
an updating stream. An alternative way is you can
use the "changelog-json" format in this repo [1], it will convert the
updating stream int
Hi Jark,
thanks for your quick reply. I was indeed expecting it.
But that triggers the following questions:
1. Is there another way to do this deduplication and generate an
append-only stream? Match Recognize? UDF? ...?
2. If I would put Postgres as a sink, what would happen? Will the e
Hi Laurent,
This is because the deduplicate node generates an updating stream, however
Kafka currently only supports append-only stream.
This can be addressed in release-1.12, because we introduce a new connector
"upsert-kafka" which supports writing updating
streams into Kafka compacted topics.
14 matches
Mail list logo
