You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/05/20 00:14:51 UTC
[GitHub] [beam] omarismail94 opened a new pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
omarismail94 opened a new pull request #11754:
URL: https://github.com/apache/beam/pull/11754
R: @TheNeuralBit
In the `BeamSqlExample.java` class, the instructions state that to run the example, use:
`./gradlew :sdks:java:extensions:sql:runBasicExample`.
I tried this and the build failed due to `java.lang.IllegalStateException: Unable to return a default Coder`
I fixed this by setting the Coder for both anon transforms.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427678033
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
I tried `setRowSchema(type)` and it failed with `: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer`.
I think it is inferring the schema as 3 fields, but the result only returns two fields, and that's why it throws the error
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427678033
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
I tried `setRowSchema(type)` and it failed with `: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer`.
I think it is inferring the schema as 3 fields, but the result only returns fields, and that's why it throws the error
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427686703
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
Actually, it is not due to the reduction in the number of field, but the order in which the fields are selected in the SELECT statement. Here is the order it expects
* Int, String, Double
and the fields that represent those types are: c1, c2, c3
If your results print out of order, it fails due to the `ClassCastException`. I tried doing this query and it failed:
`select c2, sum(c1), sum(c3) from CASE1_RESULT group by c2`,
but if I do
`select sum(c1),c2, sum(c3) from CASE1_RESULT group by c2`
it works! You can see that in the one that failed, c1 and c2s positions have switched, so the encoder trips out. What's cool is that you can see the results correctly calculated in:
` System.out.println("CASE1_RESULT: " + input.getValues());`
but it seems that when the result is encoded, the program throws an error due to the results being out of order. I guess this is because it sees `.setRowSchema(type);`, and as the order of the schema is "Int, String, Double", the results have to abide by that rule. That why it fails when we did:
`c2, sum(c3) from CASE1_RESULT group by c2`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427678727
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
This is part of the Stack trace that makes me think that
```
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer
at org.apache.beam.sdk.coders.VarIntCoder.encode(VarIntCoder.java:33)
at org.apache.beam.sdk.coders.RowCoderGenerator$EncodeInstruction.encodeDelegate(RowCoderGenerator.java:270)
at org.apache.beam.sdk.coders.Coder$ByteBuddy$E99UrF3W.encode(Unknown Source)
at org.apache.beam.sdk.coders.Coder$ByteBuddy$E99UrF3W.encode(Unknown Source)
at org.apache.beam.sdk.schemas.SchemaCoder.encode(SchemaCoder.java:115)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #11754:
URL: https://github.com/apache/beam/pull/11754#issuecomment-631654862
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #11754:
URL: https://github.com/apache/beam/pull/11754#issuecomment-631750861
I'm going to go ahead and merge since CI failure is unrelated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427674472
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
Here as well
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
Review comment:
could you change this to `withRowSchema(type)`? It does the same thing, but it's less verbose
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #11754:
URL: https://github.com/apache/beam/pull/11754#issuecomment-631665220
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #11754:
URL: https://github.com/apache/beam/pull/11754#issuecomment-631585754
retest this please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427686703
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
Actually, it is not due to the reduction in the number of field, but the order in which the fields are printed. Here is the order it expects
* Int, String, Double
and the fields that represent those types are: c1, c2, c3
If your results print out of order, it fails due to the `ClassCastException`. I tried doing this query and it failed:
`select c2, sum(c1), sum(c3) from CASE1_RESULT group by c2`,
but if I do
`select sum(c1),c2, sum(c3) from CASE1_RESULT group by c2`
it works! You can see that in the one that failed, c1 and c2s positions have switched, so the encoder trips out. What's cool is that you can see the results correctly calculated in:
` System.out.println("CASE1_RESULT: " + input.getValues());`
but it seems that when the result is encoded, the program throws an error due to the results being out of order. I guess this is because it sees `.setRowSchema(type);`, and as the order of the schema is "Int, String, Double", the results have to abide by that rule. That why it fails when we did:
`c2, sum(c3) from CASE1_RESULT group by c2`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427677681
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
Review comment:
I can do that. I did `setRowSchema(type)` and it worked!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on pull request #11754:
URL: https://github.com/apache/beam/pull/11754#issuecomment-631165775
> Thank you @omarismail94!
>
> We should probably be running this continuously to make sure we don't break it again. Would you mind adding the gradle task for this to the SQL preCommit [here](https://github.com/apache/beam/blob/master/build.gradle#L154)? That way it will run before we merge any PR that affects SQL.
I will add both this and runPojoExample to SQL preCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427681580
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
oh for this call you will need to use
```
Schema.builder()
 .addStringField("stringField")
 .addDoubleField("doubleField")
 .build()
```
like you had in the setCoder call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit merged pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
TheNeuralBit merged pull request #11754:
URL: https://github.com/apache/beam/pull/11754
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427678033
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
I tried `setRowSchema(type) and it failed with `: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer`.
I think it is inferring the schema as 3 fields, but the result only returns fields, and that's why it throws the error
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] omarismail94 commented on a change in pull request #11754: [BEAM-10037] BeamSqlExample.java fails to build
Posted by GitBox <gi...@apache.org>.
omarismail94 commented on a change in pull request #11754:
URL: https://github.com/apache/beam/pull/11754#discussion_r427686703
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/example/BeamSqlExample.java
##########
@@ -66,38 +68,47 @@ public static void main(String[] args) {
inputTable.apply(SqlTransform.query("select c1, c2, c3 from PCOLLECTION where c1 > 1"));
// print the output record of case 1;
- outputStream.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // PCOLLECTION: [3, row, 3.0]
- // PCOLLECTION: [2, row, 2.0]
- System.out.println("PCOLLECTION: " + input.getValues());
- return input;
- }
- }));
+ outputStream
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // PCOLLECTION: [3, row, 3.0]
+ // PCOLLECTION: [2, row, 2.0]
+ System.out.println("PCOLLECTION: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(RowCoder.of(type));
// Case 2. run the query with SqlTransform.query over result PCollection of case 1.
PCollection<Row> outputStream2 =
PCollectionTuple.of(new TupleTag<>("CASE1_RESULT"), outputStream)
.apply(SqlTransform.query("select c2, sum(c3) from CASE1_RESULT group by c2"));
// print the output record of case 2;
- outputStream2.apply(
- "log_result",
- MapElements.via(
- new SimpleFunction<Row, Row>() {
- @Override
- public Row apply(Row input) {
- // expect output:
- // CASE1_RESULT: [row, 5.0]
- System.out.println("CASE1_RESULT: " + input.getValues());
- return input;
- }
- }));
+ outputStream2
+ .apply(
+ "log_result",
+ MapElements.via(
+ new SimpleFunction<Row, Row>() {
+ @Override
+ public Row apply(Row input) {
+ // expect output:
+ // CASE1_RESULT: [row, 5.0]
+ System.out.println("CASE1_RESULT: " + input.getValues());
+ return input;
+ }
+ }))
+ .setCoder(
+ RowCoder.of(
+ Schema.builder()
+ .addStringField("stringField")
+ .addDoubleField("doubleField")
+ .build()));
Review comment:
Actually, it is not due to the reduction in the number of field, but the order in which the fields are printed. Here is the order it expects
* Int, String, Double
and the fields that represent those types are: c1, c2, c3
If your results print out of order, it fails due to the `ClassCastException`. I tried doing this query and it failed:
`select c2, sum(c1), sum(c3) from CASE1_RESULT group by c2`,
but if I do
`select sum(c1),c2, sum(c3) from CASE1_RESULT group by c2`
it works! You can see that in the one that failed, c1 and c2s positions have switched, so the encoder trips out. What's cool is that you can see the results correctly calculated in:
` System.out.println("CASE1_RESULT: " + input.getValues());`
but it seems that when the result is encoded, the program throws an error due to the results being out of order
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org