You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Jonathan Hope <jo...@gmail.com> on 2025/04/01 18:11:44 UTC
[Question] Redirect Write Failures to Dead Letter Queue
Hello, I had a question and was hoping this was the right place to ask.
Let's say I'm moving data from a NoSQL database to a SQL database. Here is
an example document in the NoSQL database:
{
"id": "1234",
"identifier": "5678"
}
The id is system generated, and the identifier is user provided. This is
being moved into a SQL database with two columns:
- id
- identifier
In the SQL database there is a UNIQUE index on identifier, however the same
thing cannot be enforced on the NoSQL side. Now I could check for this like
so:
1. Get source data
2. Check to see if identifier has already been inserted
3. Move duplicates to a dead letter queue
4. Write the data
5. Success
But what could happen is:
1. Get source data
2. Check to see if identifier has already been inserted
3. Move duplicates to a dead letter queue
4. Another worker inserts a duplicate identifier
5. Write the data
6. Failure
If I was doing this outside of the beam context I would try the write,
capture the errors, and then redirect the failures to some kind of dead
letter queue. However for the life of me I can't figure out how to do this
in Beam. In cases where writes are failing, retries will never succeed, and
you can't reliably check for the trigger of the failure ahead of time what
is the recommended pattern?
Thanks!
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Jonathan Hope <jo...@gmail.com>.
Of the supported Beam languages the one I know the best is Go so that's
what I've been working in. What I ended up doing was forking the databaseio
code and changing it so that any rows that had an error on a write are
emitted as a pcollection, and the error in question is also emitted as a
pcollection. I can then redirect the failed rows to a dead letter queue. I
_think_ it's working (at least it appears to locally and for small amounts
of data in Dataflow).
You are definitely right that there will likely be some trickiness when
replaying the dead letter queue. Obviously I'm hopeful that these kinds of
errors will be very infrequent.
On Thu, Apr 3, 2025 at 1:35 AM Radek Stankiewicz <ra...@google.com>
wrote:
> hi Jonathan,
> Is there a specific IO in mind you would like to use? If I think about
> enforcing key constraints, then I think about sql databases like pgsql,
> mysql, alloydb. In Beam those databases are supported in JDBCIO.
> Problem with JDBCIO is that it doesn't yet have an error handler -
> https://github.com/apache/beam/issues/20920.
> So temporarily I would recommend writing to the database from within parDo
> and handling errors like you've described.
>
> If your duplicates can happen by the same pipeline you could preprocess
> the data and partition (e.g. group by key) the data based on the
> identifier. This way the worker that is processing one identifier will
> process all the others.
> the tricky parts are when you need to solve 'at least once processing' as
> each of your entries may be reprocessed for some reason. In such a case,
> when inserting you should probably do upsert or check at least if id also
> matches and treat it as success if same (id, ident) was already inserted.
>
> On Tue, Apr 1, 2025 at 8:12 PM Jonathan Hope <
> jonathan.douglas.hope@gmail.com> wrote:
>
>> Hello, I had a question and was hoping this was the right place to ask.
>>
>> Let's say I'm moving data from a NoSQL database to a SQL database. Here
>> is an example document in the NoSQL database:
>>
>> {
>> "id": "1234",
>> "identifier": "5678"
>> }
>>
>> The id is system generated, and the identifier is user provided. This is
>> being moved into a SQL database with two columns:
>>
>> - id
>> - identifier
>>
>> In the SQL database there is a UNIQUE index on identifier, however the
>> same thing cannot be enforced on the NoSQL side. Now I could check for this
>> like so:
>>
>> 1. Get source data
>> 2. Check to see if identifier has already been inserted
>> 3. Move duplicates to a dead letter queue
>> 4. Write the data
>> 5. Success
>>
>> But what could happen is:
>>
>> 1. Get source data
>> 2. Check to see if identifier has already been inserted
>> 3. Move duplicates to a dead letter queue
>> 4. Another worker inserts a duplicate identifier
>> 5. Write the data
>> 6. Failure
>>
>>
>> If I was doing this outside of the beam context I would try the write,
>> capture the errors, and then redirect the failures to some kind of dead
>> letter queue. However for the life of me I can't figure out how to do this
>> in Beam. In cases where writes are failing, retries will never succeed, and
>> you can't reliably check for the trigger of the failure ahead of time what
>> is the recommended pattern?
>>
>> Thanks!
>>
>
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Radek Stankiewicz via user <us...@beam.apache.org>.
hi Jonathan,
Is there a specific IO in mind you would like to use? If I think about
enforcing key constraints, then I think about sql databases like pgsql,
mysql, alloydb. In Beam those databases are supported in JDBCIO.
Problem with JDBCIO is that it doesn't yet have an error handler -
https://github.com/apache/beam/issues/20920.
So temporarily I would recommend writing to the database from within parDo
and handling errors like you've described.
If your duplicates can happen by the same pipeline you could preprocess the
data and partition (e.g. group by key) the data based on the identifier.
This way the worker that is processing one identifier will process all the
others.
the tricky parts are when you need to solve 'at least once processing' as
each of your entries may be reprocessed for some reason. In such a case,
when inserting you should probably do upsert or check at least if id also
matches and treat it as success if same (id, ident) was already inserted.
On Tue, Apr 1, 2025 at 8:12 PM Jonathan Hope <
jonathan.douglas.hope@gmail.com> wrote:
> Hello, I had a question and was hoping this was the right place to ask.
>
> Let's say I'm moving data from a NoSQL database to a SQL database. Here is
> an example document in the NoSQL database:
>
> {
> "id": "1234",
> "identifier": "5678"
> }
>
> The id is system generated, and the identifier is user provided. This is
> being moved into a SQL database with two columns:
>
> - id
> - identifier
>
> In the SQL database there is a UNIQUE index on identifier, however the
> same thing cannot be enforced on the NoSQL side. Now I could check for this
> like so:
>
> 1. Get source data
> 2. Check to see if identifier has already been inserted
> 3. Move duplicates to a dead letter queue
> 4. Write the data
> 5. Success
>
> But what could happen is:
>
> 1. Get source data
> 2. Check to see if identifier has already been inserted
> 3. Move duplicates to a dead letter queue
> 4. Another worker inserts a duplicate identifier
> 5. Write the data
> 6. Failure
>
>
> If I was doing this outside of the beam context I would try the write,
> capture the errors, and then redirect the failures to some kind of dead
> letter queue. However for the life of me I can't figure out how to do this
> in Beam. In cases where writes are failing, retries will never succeed, and
> you can't reliably check for the trigger of the failure ahead of time what
> is the recommended pattern?
>
> Thanks!
>
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Robert Bradshaw via user <us...@beam.apache.org>.
On Tue, Apr 1, 2025 at 12:01 PM Jonathan Hope
<jo...@gmail.com> wrote:
>
> Unfortunately both databases will be online during this so conflicts could occur in either direction. I had previously dug up an answer around modifying the JdbcIO here: https://stackoverflow.com/questions/56398422/exception-handling-in-apache-beam-pipelines-when-writing-to-database-using-java. But I just wanted to check that there wasn't a more "official" approach that I just hadn't come across yet.
That is great. Anyone want to volunteer to move this into Beam itself?
> On Tue, Apr 1, 2025 at 11:53 AM Robert Bradshaw via user <us...@beam.apache.org> wrote:
>>
>> Good question. I think it depends on who else is modifying the SQL database.
>>
>> In the easy case (e.g. everything you want to write to your SQL
>> database comes from the NoSQL source) you could group (e.g. via a
>> GroupByKey) on your identifier, filter out duplicates with a
>> subsequent DoFn, and then write to your SQL database with the
>> assurance that there will be no duplicate keys.
>>
>> If you are concerned that there might be already existing, conflicting
>> entries in the SQL database, you can also add a step to check to see
>> if the identifier is already inserted (and, following the previous
>> step, wouldn't have to worry about another worker from this job
>> inserting the same identifier).
>>
>> If there are external, concurrent processes also modifying the SQL
>> database that you have to worry about, that gets a bit trickier, and
>> it seems necessary to do the dead letter queue as part of the write
>> itself. This may require modifying JdbcIO itself to return a dead
>> letter queue.
>>
>> - Robert
>>
>>
>>
>> On Tue, Apr 1, 2025 at 11:12 AM Jonathan Hope
>> <jo...@gmail.com> wrote:
>> >
>> > Hello, I had a question and was hoping this was the right place to ask.
>> >
>> > Let's say I'm moving data from a NoSQL database to a SQL database. Here is an example document in the NoSQL database:
>> >
>> > {
>> > "id": "1234",
>> > "identifier": "5678"
>> > }
>> >
>> > The id is system generated, and the identifier is user provided. This is being moved into a SQL database with two columns:
>> >
>> > id
>> > identifier
>> >
>> > In the SQL database there is a UNIQUE index on identifier, however the same thing cannot be enforced on the NoSQL side. Now I could check for this like so:
>> >
>> > Get source data
>> > Check to see if identifier has already been inserted
>> > Move duplicates to a dead letter queue
>> > Write the data
>> > Success
>> >
>> > But what could happen is:
>> >
>> > Get source data
>> > Check to see if identifier has already been inserted
>> > Move duplicates to a dead letter queue
>> > Another worker inserts a duplicate identifier
>> > Write the data
>> > Failure
>> >
>> >
>> > If I was doing this outside of the beam context I would try the write, capture the errors, and then redirect the failures to some kind of dead letter queue. However for the life of me I can't figure out how to do this in Beam. In cases where writes are failing, retries will never succeed, and you can't reliably check for the trigger of the failure ahead of time what is the recommended pattern?
>> >
>> > Thanks!
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Robert Bradshaw via dev <de...@beam.apache.org>.
On Tue, Apr 1, 2025 at 12:01 PM Jonathan Hope
<jo...@gmail.com> wrote:
>
> Unfortunately both databases will be online during this so conflicts could occur in either direction. I had previously dug up an answer around modifying the JdbcIO here: https://stackoverflow.com/questions/56398422/exception-handling-in-apache-beam-pipelines-when-writing-to-database-using-java. But I just wanted to check that there wasn't a more "official" approach that I just hadn't come across yet.
That is great. Anyone want to volunteer to move this into Beam itself?
> On Tue, Apr 1, 2025 at 11:53 AM Robert Bradshaw via user <us...@beam.apache.org> wrote:
>>
>> Good question. I think it depends on who else is modifying the SQL database.
>>
>> In the easy case (e.g. everything you want to write to your SQL
>> database comes from the NoSQL source) you could group (e.g. via a
>> GroupByKey) on your identifier, filter out duplicates with a
>> subsequent DoFn, and then write to your SQL database with the
>> assurance that there will be no duplicate keys.
>>
>> If you are concerned that there might be already existing, conflicting
>> entries in the SQL database, you can also add a step to check to see
>> if the identifier is already inserted (and, following the previous
>> step, wouldn't have to worry about another worker from this job
>> inserting the same identifier).
>>
>> If there are external, concurrent processes also modifying the SQL
>> database that you have to worry about, that gets a bit trickier, and
>> it seems necessary to do the dead letter queue as part of the write
>> itself. This may require modifying JdbcIO itself to return a dead
>> letter queue.
>>
>> - Robert
>>
>>
>>
>> On Tue, Apr 1, 2025 at 11:12 AM Jonathan Hope
>> <jo...@gmail.com> wrote:
>> >
>> > Hello, I had a question and was hoping this was the right place to ask.
>> >
>> > Let's say I'm moving data from a NoSQL database to a SQL database. Here is an example document in the NoSQL database:
>> >
>> > {
>> > "id": "1234",
>> > "identifier": "5678"
>> > }
>> >
>> > The id is system generated, and the identifier is user provided. This is being moved into a SQL database with two columns:
>> >
>> > id
>> > identifier
>> >
>> > In the SQL database there is a UNIQUE index on identifier, however the same thing cannot be enforced on the NoSQL side. Now I could check for this like so:
>> >
>> > Get source data
>> > Check to see if identifier has already been inserted
>> > Move duplicates to a dead letter queue
>> > Write the data
>> > Success
>> >
>> > But what could happen is:
>> >
>> > Get source data
>> > Check to see if identifier has already been inserted
>> > Move duplicates to a dead letter queue
>> > Another worker inserts a duplicate identifier
>> > Write the data
>> > Failure
>> >
>> >
>> > If I was doing this outside of the beam context I would try the write, capture the errors, and then redirect the failures to some kind of dead letter queue. However for the life of me I can't figure out how to do this in Beam. In cases where writes are failing, retries will never succeed, and you can't reliably check for the trigger of the failure ahead of time what is the recommended pattern?
>> >
>> > Thanks!
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Jonathan Hope <jo...@gmail.com>.
Unfortunately both databases will be online during this so conflicts could
occur in either direction. I had previously dug up an answer around
modifying the JdbcIO here:
https://stackoverflow.com/questions/56398422/exception-handling-in-apache-beam-pipelines-when-writing-to-database-using-java.
But I just wanted to check that there wasn't a more "official" approach
that I just hadn't come across yet.
On Tue, Apr 1, 2025 at 11:53 AM Robert Bradshaw via user <
user@beam.apache.org> wrote:
> Good question. I think it depends on who else is modifying the SQL
> database.
>
> In the easy case (e.g. everything you want to write to your SQL
> database comes from the NoSQL source) you could group (e.g. via a
> GroupByKey) on your identifier, filter out duplicates with a
> subsequent DoFn, and then write to your SQL database with the
> assurance that there will be no duplicate keys.
>
> If you are concerned that there might be already existing, conflicting
> entries in the SQL database, you can also add a step to check to see
> if the identifier is already inserted (and, following the previous
> step, wouldn't have to worry about another worker from this job
> inserting the same identifier).
>
> If there are external, concurrent processes also modifying the SQL
> database that you have to worry about, that gets a bit trickier, and
> it seems necessary to do the dead letter queue as part of the write
> itself. This may require modifying JdbcIO itself to return a dead
> letter queue.
>
> - Robert
>
>
>
> On Tue, Apr 1, 2025 at 11:12 AM Jonathan Hope
> <jo...@gmail.com> wrote:
> >
> > Hello, I had a question and was hoping this was the right place to ask.
> >
> > Let's say I'm moving data from a NoSQL database to a SQL database. Here
> is an example document in the NoSQL database:
> >
> > {
> > "id": "1234",
> > "identifier": "5678"
> > }
> >
> > The id is system generated, and the identifier is user provided. This is
> being moved into a SQL database with two columns:
> >
> > id
> > identifier
> >
> > In the SQL database there is a UNIQUE index on identifier, however the
> same thing cannot be enforced on the NoSQL side. Now I could check for this
> like so:
> >
> > Get source data
> > Check to see if identifier has already been inserted
> > Move duplicates to a dead letter queue
> > Write the data
> > Success
> >
> > But what could happen is:
> >
> > Get source data
> > Check to see if identifier has already been inserted
> > Move duplicates to a dead letter queue
> > Another worker inserts a duplicate identifier
> > Write the data
> > Failure
> >
> >
> > If I was doing this outside of the beam context I would try the write,
> capture the errors, and then redirect the failures to some kind of dead
> letter queue. However for the life of me I can't figure out how to do this
> in Beam. In cases where writes are failing, retries will never succeed, and
> you can't reliably check for the trigger of the failure ahead of time what
> is the recommended pattern?
> >
> > Thanks!
>
Re: [Question] Redirect Write Failures to Dead Letter Queue
Posted by Robert Bradshaw via user <us...@beam.apache.org>.
Good question. I think it depends on who else is modifying the SQL database.
In the easy case (e.g. everything you want to write to your SQL
database comes from the NoSQL source) you could group (e.g. via a
GroupByKey) on your identifier, filter out duplicates with a
subsequent DoFn, and then write to your SQL database with the
assurance that there will be no duplicate keys.
If you are concerned that there might be already existing, conflicting
entries in the SQL database, you can also add a step to check to see
if the identifier is already inserted (and, following the previous
step, wouldn't have to worry about another worker from this job
inserting the same identifier).
If there are external, concurrent processes also modifying the SQL
database that you have to worry about, that gets a bit trickier, and
it seems necessary to do the dead letter queue as part of the write
itself. This may require modifying JdbcIO itself to return a dead
letter queue.
- Robert
On Tue, Apr 1, 2025 at 11:12 AM Jonathan Hope
<jo...@gmail.com> wrote:
>
> Hello, I had a question and was hoping this was the right place to ask.
>
> Let's say I'm moving data from a NoSQL database to a SQL database. Here is an example document in the NoSQL database:
>
> {
> "id": "1234",
> "identifier": "5678"
> }
>
> The id is system generated, and the identifier is user provided. This is being moved into a SQL database with two columns:
>
> id
> identifier
>
> In the SQL database there is a UNIQUE index on identifier, however the same thing cannot be enforced on the NoSQL side. Now I could check for this like so:
>
> Get source data
> Check to see if identifier has already been inserted
> Move duplicates to a dead letter queue
> Write the data
> Success
>
> But what could happen is:
>
> Get source data
> Check to see if identifier has already been inserted
> Move duplicates to a dead letter queue
> Another worker inserts a duplicate identifier
> Write the data
> Failure
>
>
> If I was doing this outside of the beam context I would try the write, capture the errors, and then redirect the failures to some kind of dead letter queue. However for the life of me I can't figure out how to do this in Beam. In cases where writes are failing, retries will never succeed, and you can't reliably check for the trigger of the failure ahead of time what is the recommended pattern?
>
> Thanks!