Mixing
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
794722
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
794722
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
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
2 days ago
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
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago
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
794722
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
794722
asked Nov 20 '18 at 17:35
salientsalient
794722
asked Nov 20 '18 at 17:35
salientsalient
794722
asked Nov 20 '18 at 17:35
salientsalient
794722
asked Nov 20 '18 at 17:35
salientsalient
794722
794722
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
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
2 days ago
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
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago
add a comment |
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
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
2 days ago
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
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
2 days ago
This question would be more answerable if you could specify the desired balance of latency versus cost.
– Kenn Knowles
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
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
2 days ago
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
2 days ago
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
2 days ago
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
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago
add a comment |
1 Answer
1
active
oldest
votes
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 2 days ago
Rui WangRui Wang
715
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
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
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 2 days ago
Rui WangRui Wang
715
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
add a comment |
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 2 days ago
Rui WangRui Wang
715
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
add a comment |
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 2 days ago
Rui WangRui Wang
715
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 2 days ago
Rui WangRui Wang
715
edited 2 days ago
answered 2 days ago
Rui WangRui Wang
715
answered 2 days ago
Rui WangRui Wang
715
answered 2 days ago
Rui WangRui Wang
715
715
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
add a comment |
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
The join lib assumes that the keys are equal — they are not in my example.
– salient
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
@salient Do you join Person.OrganizationID on Organization.ID?
– Rui Wang
yesterday
yes that is exactly so
– salient
yesterday
yes that is exactly so
– salient
yesterday
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
yesterday
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
yesterday
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
2 days ago
@KennKnowles — latency is not so important, so long as a new Persons are always emitted for each change in Organization.
– salient
2 days ago
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
2 days ago
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
2 days ago
Given unlimited storage, my answer will apply then.
– Rui Wang
2 days ago