Idiomatic way to join on “secondary” keys
If we have a stream that looks like this
Person {
…
OrganizationID
}
that we want to join with another stream
Organization {
ID
…
}
to create a composite record like so:
Person {
…
Organization {
ID
…
}
}
What is the most idiomatic and efficient way to do so in the Apache Beam programming model?
NB: have seen side inputs recommended as a solution to similar problems like this, but it is not applicable here since the effect we are after is that every change to either Person or Organization should yield a new augmented Person-record.
apache-beam
asked Nov 20 '18 at 17:35
salientsalient
799722
add a comment |
If we have a stream that looks like this
Person {
…
OrganizationID
}
that we want to join with another stream
Organization {
ID
…
}
to create a composite record like so:
Person {
…
Organization {
ID
…
}
}
What is the most idiomatic and efficient way to do so in the Apache Beam programming model?
NB: have seen side inputs recommended as a solution to similar problems like this, but it is not applicable here since the effect we are after is that every change to either Person or Organization should yield a new augmented Person-record.
apache-beam
asked Nov 20 '18 at 17:35
salientsalient
799722
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49
add a comment |
If we have a stream that looks like this
Person {
…
OrganizationID
}
that we want to join with another stream
Organization {
ID
…
}
to create a composite record like so:
Person {
…
Organization {
ID
…
}
}
What is the most idiomatic and efficient way to do so in the Apache Beam programming model?
NB: have seen side inputs recommended as a solution to similar problems like this, but it is not applicable here since the effect we are after is that every change to either Person or Organization should yield a new augmented Person-record.
apache-beam
asked Nov 20 '18 at 17:35
salientsalient
799722
If we have a stream that looks like this
Person {
…
OrganizationID
}
that we want to join with another stream
Organization {
ID
…
}
to create a composite record like so:
Person {
…
Organization {
ID
…
}
}
What is the most idiomatic and efficient way to do so in the Apache Beam programming model?
NB: have seen side inputs recommended as a solution to similar problems like this, but it is not applicable here since the effect we are after is that every change to either Person or Organization should yield a new augmented Person-record.
apache-beam
apache-beam
asked Nov 20 '18 at 17:35
salientsalient
799722
asked Nov 20 '18 at 17:35
salientsalient
799722
asked Nov 20 '18 at 17:35
salientsalient
799722
asked Nov 20 '18 at 17:35
salientsalient
799722
asked Nov 20 '18 at 17:35
salientsalient
799722
799722
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49
add a comment |
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49
add a comment |
1 Answer
1
active
oldest
votes
Edit:
The answer is, your example is not supported by Apache Beam due to missing retraction in Apache Beam implementation.
===================================================
Original answer:
You might want to check Join library[1] in Apache Beam.
Join in Beam model needs extra thinking on windowing strategies on your streams. Sounds like your streams does not require windowing, so say your streams are both in global window. But if you set global window on both of your streams, use default trigger and do Join like Beam's Join library, due to watermark never passes endless window, your Join will not emit any result. If you set repeatly data driven trigger (fire once seen enough elements), however, due to missing supporting for retraction in Beam, it's not clear how pre-emited result is refined for Join.
[1] https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L49
answered Jan 25 at 21:59
Rui WangRui Wang
14218
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
add a comment |
Your Answer
StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");
StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});
function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});
}
});
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53398511%2fidiomatic-way-to-join-on-secondary-keys%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
1 Answer
1
active
oldest
votes
1 Answer
1
active
oldest
votes
active
oldest
votes
active
oldest
votes
Edit:
The answer is, your example is not supported by Apache Beam due to missing retraction in Apache Beam implementation.
===================================================
Original answer:
You might want to check Join library[1] in Apache Beam.
Join in Beam model needs extra thinking on windowing strategies on your streams. Sounds like your streams does not require windowing, so say your streams are both in global window. But if you set global window on both of your streams, use default trigger and do Join like Beam's Join library, due to watermark never passes endless window, your Join will not emit any result. If you set repeatly data driven trigger (fire once seen enough elements), however, due to missing supporting for retraction in Beam, it's not clear how pre-emited result is refined for Join.
[1] https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L49
answered Jan 25 at 21:59
Rui WangRui Wang
14218
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
add a comment |
Edit:
The answer is, your example is not supported by Apache Beam due to missing retraction in Apache Beam implementation.
===================================================
Original answer:
You might want to check Join library[1] in Apache Beam.
Join in Beam model needs extra thinking on windowing strategies on your streams. Sounds like your streams does not require windowing, so say your streams are both in global window. But if you set global window on both of your streams, use default trigger and do Join like Beam's Join library, due to watermark never passes endless window, your Join will not emit any result. If you set repeatly data driven trigger (fire once seen enough elements), however, due to missing supporting for retraction in Beam, it's not clear how pre-emited result is refined for Join.
[1] https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L49
answered Jan 25 at 21:59
Rui WangRui Wang
14218
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
add a comment |
Edit:
The answer is, your example is not supported by Apache Beam due to missing retraction in Apache Beam implementation.
===================================================
Original answer:
You might want to check Join library[1] in Apache Beam.
Join in Beam model needs extra thinking on windowing strategies on your streams. Sounds like your streams does not require windowing, so say your streams are both in global window. But if you set global window on both of your streams, use default trigger and do Join like Beam's Join library, due to watermark never passes endless window, your Join will not emit any result. If you set repeatly data driven trigger (fire once seen enough elements), however, due to missing supporting for retraction in Beam, it's not clear how pre-emited result is refined for Join.
[1] https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L49
answered Jan 25 at 21:59
Rui WangRui Wang
14218
Edit:
The answer is, your example is not supported by Apache Beam due to missing retraction in Apache Beam implementation.
===================================================
Original answer:
You might want to check Join library[1] in Apache Beam.
Join in Beam model needs extra thinking on windowing strategies on your streams. Sounds like your streams does not require windowing, so say your streams are both in global window. But if you set global window on both of your streams, use default trigger and do Join like Beam's Join library, due to watermark never passes endless window, your Join will not emit any result. If you set repeatly data driven trigger (fire once seen enough elements), however, due to missing supporting for retraction in Beam, it's not clear how pre-emited result is refined for Join.
[1] https://github.com/apache/beam/blob/master/sdks/java/extensions/join-library/src/main/java/org/apache/beam/sdk/extensions/joinlibrary/Join.java#L49
answered Jan 25 at 21:59
Rui WangRui Wang
14218
edited Jan 28 at 22:34
answered Jan 25 at 21:59
Rui WangRui Wang
14218
answered Jan 25 at 21:59
Rui WangRui Wang
14218
answered Jan 25 at 21:59
Rui WangRui Wang
14218
14218
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
add a comment |
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
The join lib assumes that the keys are equal — they are not in my example.
– salient
Jan 26 at 9:42
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
Jan 27 at 4:14
yes that is exactly so
– salient
Jan 27 at 4:15
yes that is exactly so
– salient
Jan 27 at 4:15
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
Then you can have PCollections of KV<OrganizationID, Person> and KV<ID, Organization> and then join them together. The Key in Join library is a type, as long as your ID in Person and Org are same type, they can be joined: github.com/apache/beam/blob/master/sdks/java/extensions/…
– Rui Wang
Jan 27 at 4:20
add a comment |
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53398511%2fidiomatic-way-to-join-on-secondary-keys%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
Jan 25 at 22:13
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
Jan 25 at 22:19
How about the cost? Conceptually if you do not specify cost limitation, all elements will be buffered from both stream. Say if there is a new org id appear on org stream, all person with the same org id is supposed to be buffered at that moment and then emits all new joined elements (of course some optimization can be done. e.g drop person events that has been joined)
– Rui Wang
Jan 25 at 22:36
Yes, I understand that that would have to be the case. But lets assume Person and Organization are ”fairly small” (N ~1-100M records). Anyway would be interesting to hear the different options based on different answers to the cost question.
– salient
Jan 25 at 22:41
Given unlimited storage, my answer will apply then.
– Rui Wang
Jan 26 at 4:49