[BEAM-4461] A transform to perform binary joins of PCollections with schemas #8273

reuvenlax · 2019-04-11T00:22:04Z

This supports inner, outer, left outer, and right outer joins.

Example usage:

pCollection1.apply(Join.innerJoin(pCollection2).using("user", "country"));

robinyqiu

Some idea on the API design of CoGroup and Join:
Currently we have

pcTuple.apply(CoGroup.join(...).join(...)) output a PCollection<KV<Row, Row>>
pcTuple.apply(CoGroup.join(...).join(...).crossProductJoin(...)) output a PCollection<Row>
pc1.apply(Join.innerJoin(pc2).on(...)) output a PCollection<Row>

From a user's perspective, I would say 2) and 3) share more commonality than 1) and 2), because they both output the schema of a cross product of all/both inputs; whereas 1) output a KV<K, map from tag to Iterable>. So how about we move the implementation of 2) under the Join class?

Also, with the crossProductJoin() removed, CoGroup will always output PCollection<KV<Row, Row>>, which makes its usage closer to Group.ByFields. (They both do aggregation on one table/multiple tables, instead of cross product.)

Please let me know what you think. I am also happy to help with making some changes if needed.

robinyqiu · 2019-04-11T18:52:47Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/CoGroup.java

@@ -159,7 +159,7 @@
 *
 * <pre>{@code
 * PCollection<Row> joined = PCollectionTuple.of("input1", input1, "input2", input2)
- * .apply(CoGroup.join("input1", By.fieldNames("user").withOuterJoinParticipation())
+ * .apply(CoGroup.join("input1", By.fieldNames("user").withOptionalParticipation())


Nit: update doc at L152 as well?

robinyqiu · 2019-04-11T22:15:17Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/CoGroup.java

@@ -244,7 +244,7 @@ public static By fieldAccessDescriptor(FieldAccessDescriptor fieldAccessDescript
 *
 * <p>This only affects the results of expandCrossProduct.
 */
- public By withOuterJoinParticipation() {
+ public By withOptionalParticipation() {
 return toBuilder().setOuterJoinParticipation(true).build();


I think this should be changed to get/setOptionalParticipation()

robinyqiu · 2019-04-11T23:49:17Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Join.java

+ *
+ * <p>Full outer joins, left outer joins, and right outer joins are also supported.
+ */
+public class Join {


Shouldn't we mark this class and CoGroup with @Experimental(Kind.SCHEMAS)? I think we may be able to make some improvement on the API design later.

good catch! done.

robinyqiu · 2019-04-12T17:54:33Z

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/JoinTest.java

+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testRightOuterJoin() {


Nit: rename to testRightOuterJoinSameKeys?

robinyqiu · 2019-04-12T17:55:00Z

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/JoinTest.java

+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testLeftOuterJoin() {


Nit: rename to testLeftOuterJoinSameKeys?

robinyqiu · 2019-04-12T21:21:05Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Join.java

+ };
+
+ /** Implementation class . */
+ public static class Inner<LhsT, RhsT> extends PTransform<PCollection<LhsT>, PCollection<Row>> {


Nit: I personally find the name Inner a bit misleading because we also have "inner" joins here, but I saw this name is used throughout the package. Maybe something like "Impl" is more unambiguous.

I used Inner, because that's the name traditionally used in BeamJava transforms. However I agree it is confusing in this context; changed to Impl.

robinyqiu · 2019-04-12T22:08:47Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Join.java

+ }
+
+ @Override
+ public PCollection<Row> expand(PCollection lhs) {


Why don't use the more specific type here: PCollection<LhsT>

Because the output schema here is the joined schema, so it would not match the schema of either LhsT or RhsT.

I was asking about the input parameter PCollection lhs, not the return type.

robinyqiu · 2019-04-16T17:34:59Z

LGTM. Also please let me know what's your opinion on my comments here: #8273 (review)

reuvenlax · 2019-04-17T05:21:24Z

@robinyqiu I think the fundamental difference between CoGroup and Join isn't the cross product, it's that CoGroup is a general grouping/join on N inputs while Join is something closer to a standard binary join. Moving the cross-product into Join would mean that we would lose it for the N-input case where N > 2. I don't think we want the Join transform to start dealing with PCollectionTuples - the whole point of the binary join transform is simply to be syntactic sugar that makes common use cases easy.

You are .correct that standard CoGroup is closer to Group. That's the same with standard Beam too - CoGroupByKey is just the multi-input version of GroupByKey. I think that it would make sense to allow similar aggregations in CoGroup as we do in Group

reuvenlax · 2019-04-17T14:12:18Z

@robinyqiu I also considered calling this transform BinaryJoin to emphasize that, but opted for the shorter name (since SQL joins are always binary joins anyway).

robinyqiu · 2019-04-17T18:02:53Z

@robinyqiu I think the fundamental difference between CoGroup and Join isn't the cross product, it's that CoGroup is a general grouping/join on N inputs while Join is something closer to a standard binary join.

Yes I agree that's the fundamental difference between the current design of CoGroup and Join

Moving the cross-product into Join would mean that we would lose it for the N-input case where N > 2.

I was thinking of providing two different flavors of APIs and put them both under Join, one for the general N case (deal with PCollectionTuple) and one for N=2 case (syntactic sugar for the first case).

I don't think we want the Join transform to start dealing with PCollectionTuples - the whole point of the binary join transform is simply to be syntactic sugar that makes common use cases easy.

Ok I see your point. If binary join is a much more common use case than N>2 join, then I agree that the current API is easier to use.

reuvenlax · 2019-04-18T01:40:42Z

Ah sorry I missed that! I'll send another PR to fix the lhs parameter.

…

On Wed, Apr 17, 2019 at 5:04 PM Yueyang Qiu ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Join.java <#8273 (comment)>: > + /** + * Perform a natural join between the PCollections. The fields are expected to exist in both + * PCollections + */ + public Inner<LhsT, RhsT> using(FieldAccessDescriptor fieldAccessDescriptor) { + return new Inner<>( + joinType, rhs, FieldsEqual.left(fieldAccessDescriptor).right(fieldAccessDescriptor)); + } + + /** Join the PCollections using the provided predicate. */ + public Inner<LhsT, RhsT> on(FieldsEqual.Inner predicate) { + return new Inner<>(joinType, rhs, predicate); + } + + @OverRide + public PCollection<Row> expand(PCollection lhs) { I was asking about the input parameter PCollection lhs, not the return type. — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#8273 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AFAYJVKAK4MQ4IJNN5MMZCTPQ63KRANCNFSM4HFB7O6A> .

…nary joins of PCollections with schemas

Add a binary join API

e8d96da

reuvenlax force-pushed the schema_join branch from 563c20a to e8d96da Compare April 11, 2019 07:13

Fix Javadoc.

fcae574

robinyqiu approved these changes Apr 12, 2019

View reviewed changes

Address comments.

1718a59

Fix remaining issues

d633dcc

reuvenlax merged commit a00ba2b into apache:master Apr 17, 2019

ibzib pushed a commit to ibzib/beam that referenced this pull request Apr 22, 2019

Merge pull request apache#8273: [BEAM-4461] A transform to perform bi…

527633e

…nary joins of PCollections with schemas

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-4461] A transform to perform binary joins of PCollections with schemas #8273

[BEAM-4461] A transform to perform binary joins of PCollections with schemas #8273

reuvenlax commented Apr 11, 2019

robinyqiu left a comment •

edited

Loading

robinyqiu Apr 11, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 11, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 11, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 12, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 12, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 12, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 12, 2019

reuvenlax Apr 16, 2019

robinyqiu Apr 18, 2019

robinyqiu commented Apr 16, 2019

reuvenlax commented Apr 17, 2019

reuvenlax commented Apr 17, 2019

robinyqiu commented Apr 17, 2019 •

edited

Loading

reuvenlax commented Apr 18, 2019 via email

[BEAM-4461] A transform to perform binary joins of PCollections with schemas #8273

[BEAM-4461] A transform to perform binary joins of PCollections with schemas #8273

Conversation

reuvenlax commented Apr 11, 2019

robinyqiu left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robinyqiu commented Apr 16, 2019

reuvenlax commented Apr 17, 2019

reuvenlax commented Apr 17, 2019

robinyqiu commented Apr 17, 2019 • edited Loading

reuvenlax commented Apr 18, 2019 via email

robinyqiu left a comment •

edited

Loading

robinyqiu commented Apr 17, 2019 •

edited

Loading