Re: Apache Beam Go SDK Quickstart bugs
Awesome! Just going to add a few colleagues (who are subscribed anyhow) to make sure this hits the top of their inbox. +Robert Burke +Chamikara Jayalath +Kyle Weaver Kenn On Thu, Nov 4, 2021 at 11:40 PM Jeff Rhyason wrote: > I'm interested to see the Go SDK work with the Spark runner. Based on the > instructions at https://beam.apache.org/get-started/quickstart-go/, I run > these commands and get the following failure: > > $ ./gradlew :runners:spark:2:job-server:runShadow > in another window: > $ cd sdks > $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo --runner > spark --endpoint localhost:8099 > 2021/11/04 22:06:46 No environment config specified. Using default config: > 'apache/beam_go_sdk:2.35.0.dev' > 2021/11/04 22:06:46 Failed to execute job: generating model pipeline > failed to add scope tree: &{{CountWords root/CountWords} [{main.extractFn > 5: ParDo [In(Main): string <- {4: string/string GLO}] -> [Out: string -> > {5: string/string GLO}]}] [0xc96cd0]} > caused by: > failed to add input kind: {main.extractFn 5: ParDo [In(Main): string <- > {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}]} > caused by: > failed to serialize 5: ParDo [In(Main): string <- {4: string/string GLO}] > -> [Out: string -> {5: string/string GLO}] > caused by: > encoding userfn 5: ParDo [In(Main): string <- {4: string/string > GLO}] -> [Out: string -> {5: string/string GLO}] > bad userfn > caused by: > encoding structural DoFn &{ 0xc000460ae8 > map[ProcessElement:0xc0004fcac0] map[]} > receiver type *main.extractFn must be registered > exit status 1 > > I was able to register that type, like this: > > diff --git a/sdks/go/examples/wordcount/wordcount.go > b/sdks/go/examples/wordcount/wordcount.go > index 4d54db9a2d..6db99d6220 100644 > --- a/sdks/go/examples/wordcount/wordcount.go > +++ b/sdks/go/examples/wordcount/wordcount.go > @@ -60,6 +60,7 @@ import ( > "flag" > "fmt" > "log" > + "reflect" > "regexp" > "strings" > > @@ -107,6 +108,7 @@ var ( > // by calling beam.RegisterFunction in an init() call. > func init() { > beam.RegisterFunction(formatFn) > + beam.RegisterDoFn(reflect.TypeOf((*extractFn)(nil)).Elem()) > } > > var ( > > > Then I encountered: > > $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo --runner > spark --endpoint localhost:8099 > ... > 2021/11/04 23:07:26 (): java.lang.IllegalArgumentException: Unsupported > class file major version 55 > 2021/11/04 23:07:26 Job state: FAILED > 2021/11/04 23:07:26 Failed to execute job: job > go0job0101636092444228385176-jar-1105060726-caffd2f4_ef37c3a9-b2a8-47bd-b1c7-a6e2771263f2 > failed > exit status 1 > > > Switching to the Spark 3.0 job server changed things: > $ cd .. ; ./gradlew :runners:spark:3:job-server:runShadow > ... > $ cd sdks ; go1.16.4 run ./go/examples/wordcount/wordcount.go --output > foo --runner spark --endpoint localhost:8099 > ... > 2021/11/04 23:12:04 Staged binary artifact with token: > 2021/11/04 23:12:04 Submitted job: > go0job0101636092722590274154-jar-1105061204-7f1d879e_28e28e06-1331-41c6-8288-4dcfa87afd13 > 2021/11/04 23:12:04 Job state: STOPPED > 2021/11/04 23:12:04 Job state: STARTING > 2021/11/04 23:12:04 Job state: RUNNING > 2021/11/04 23:12:17 Job state: DONE > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x01" labels:{key:"PCOLLECTION" value:"n1"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x01" labels:{key:"PCOLLECTION" value:"n3"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x81\xdc\x01" labels:{key:"PCOLLECTION" value:"n5"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x01" labels:{key:"PCOLLECTION" value:"n2"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x95+" labels:{key:"PCOLLECTION" value:"n4"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:element_count:v1" type:"beam:metrics:sum_int64:v1" > payload:"\x81\xdc\x01" labels:{key:"PCOLLECTION" value:"n6"} > 2021/11/04 23:12:17 unknown metric type > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:sampled_byte_size:v1" > type:"beam:metrics:distribution_int64:v1" payload:"\x01\x01\x01\x01" > labels:{key:"PCOLLECTION" value:"n1"} > 2021/11/04 23:12:17 Failed to deduce Step from MonitoringInfo: > urn:"beam:metric:sampled_byte_size:v1" > type:"beam:metrics:distribution_int64:v1" payload:"\x01222" > labels:
Re: Apache Beam Go SDK Quickstart bugs
With the 2.35.0 cut coming on Wednesday (Nov 17th) I took the liberty to fix all the Go SDK examples under Spark 3. I don't like "stealing" work, but we had not heard from you since this was brought to our attention. So, for that, I'm sorry. https://github.com/apache/beam/pull/15970 Found a bug with the schema row decoder along the way too. Since the website tracks live, getting the quick start to use Spark 3 doesn't have to happen before the cut, so that's still available to do. I really appreciate the clear errors and repros you provided! Thanks again Robert Burke A On Mon, Nov 8, 2021, 1:27 PM Robert Burke wrote: > +1 to Kyle's LOOPBACK suggestion. Gives you your local file system, and > you can println debug to the console. However, only would be a single > worker. > > On Mon, Nov 8, 2021 at 1:23 PM Robert Burke wrote: > >> Oh that's definitely something needs updating. Yes please to those PRs. >> >> Please add a mention to @lostluck for me to review them. >> >> The "Unsupported class file major version" is a mismatch between Java8 >> and Java 11, unrelated to the Go SDK, so I agree that the example should >> spin up a spark3 instead of the older version. >> >> The "Failed to deduce Step from MonitoringInfo" messages are an >> unfortunately noisy error message post successful job, because the code >> doesn't know how to map PCollections to their Parent DoFn yet. Ritesh is >> working on that. They probably need to be consolidated or ignored for now. >> Right now, they come from here: >> https://github.com/apache/beam/blob/e668460f61540638fb29e05997087b56ebcee4f3/sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go#L51 >> >> >> >> On Mon, Nov 8, 2021 at 1:12 PM Kenneth Knowles wrote: >> >>> Awesome! Just going to add a few colleagues (who are subscribed anyhow) >>> to make sure this hits the top of their inbox. >>> >>> +Robert Burke +Chamikara Jayalath >>> +Kyle Weaver >>> >>> Kenn >>> >>> On Thu, Nov 4, 2021 at 11:40 PM Jeff Rhyason wrote: >>> I'm interested to see the Go SDK work with the Spark runner. Based on the instructions at https://beam.apache.org/get-started/quickstart-go/, I run these commands and get the following failure: $ ./gradlew :runners:spark:2:job-server:runShadow in another window: $ cd sdks $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo --runner spark --endpoint localhost:8099 2021/11/04 22:06:46 No environment config specified. Using default config: 'apache/beam_go_sdk:2.35.0.dev' 2021/11/04 22:06:46 Failed to execute job: generating model pipeline failed to add scope tree: &{{CountWords root/CountWords} [{main.extractFn 5: ParDo [In(Main): string <- {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}]}] [0xc96cd0]} caused by: failed to add input kind: {main.extractFn 5: ParDo [In(Main): string <- {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}]} caused by: failed to serialize 5: ParDo [In(Main): string <- {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}] caused by: encoding userfn 5: ParDo [In(Main): string <- {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}] bad userfn caused by: encoding structural DoFn &{ 0xc000460ae8 map[ProcessElement:0xc0004fcac0] map[]} receiver type *main.extractFn must be registered exit status 1 I was able to register that type, like this: diff --git a/sdks/go/examples/wordcount/wordcount.go b/sdks/go/examples/wordcount/wordcount.go index 4d54db9a2d..6db99d6220 100644 --- a/sdks/go/examples/wordcount/wordcount.go +++ b/sdks/go/examples/wordcount/wordcount.go @@ -60,6 +60,7 @@ import ( "flag" "fmt" "log" + "reflect" "regexp" "strings" @@ -107,6 +108,7 @@ var ( // by calling beam.RegisterFunction in an init() call. func init() { beam.RegisterFunction(formatFn) + beam.RegisterDoFn(reflect.TypeOf((*extractFn)(nil)).Elem()) } var ( Then I encountered: $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo --runner spark --endpoint localhost:8099 ... 2021/11/04 23:07:26 (): java.lang.IllegalArgumentException: Unsupported class file major version 55 2021/11/04 23:07:26 Job state: FAILED 2021/11/04 23:07:26 Failed to execute job: job go0job0101636092444228385176-jar-1105060726-caffd2f4_ef37c3a9-b2a8-47bd-b1c7-a6e2771263f2 failed exit status 1 Switching to the Spark 3.0 job server changed things: $ cd .. ; ./gradlew :runners:spark:3:job-server:runShadow ... $ cd sdks ; go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo --runner spark --endpoint
Re: Apache Beam Go SDK Quickstart bugs
Hi Robert! Your fix was much more thorough than what I was cooking up. Thanks for such a comprehensive fix. Also, thanks to you and Kyle for explaining the details of the other things I observed. This is working very well now! Jeff On Sun, Nov 14, 2021 at 4:04 PM Robert Burke wrote: > With the 2.35.0 cut coming on Wednesday (Nov 17th) I took the liberty to > fix all the Go SDK examples under Spark 3. I don't like "stealing" work, > but we had not heard from you since this was brought to our attention. So, > for that, I'm sorry. > > https://github.com/apache/beam/pull/15970 > > Found a bug with the schema row decoder along the way too. > > Since the website tracks live, getting the quick start to use Spark 3 > doesn't have to happen before the cut, so that's still available to do. > > I really appreciate the clear errors and repros you provided! > > Thanks again > Robert Burke > > A > > On Mon, Nov 8, 2021, 1:27 PM Robert Burke wrote: > >> +1 to Kyle's LOOPBACK suggestion. Gives you your local file system, and >> you can println debug to the console. However, only would be a single >> worker. >> >> On Mon, Nov 8, 2021 at 1:23 PM Robert Burke wrote: >> >>> Oh that's definitely something needs updating. Yes please to those PRs. >>> >>> Please add a mention to @lostluck for me to review them. >>> >>> The "Unsupported class file major version" is a mismatch between Java8 >>> and Java 11, unrelated to the Go SDK, so I agree that the example should >>> spin up a spark3 instead of the older version. >>> >>> The "Failed to deduce Step from MonitoringInfo" messages are an >>> unfortunately noisy error message post successful job, because the code >>> doesn't know how to map PCollections to their Parent DoFn yet. Ritesh is >>> working on that. They probably need to be consolidated or ignored for now. >>> Right now, they come from here: >>> https://github.com/apache/beam/blob/e668460f61540638fb29e05997087b56ebcee4f3/sdks/go/pkg/beam/core/runtime/metricsx/metricsx.go#L51 >>> >>> >>> >>> On Mon, Nov 8, 2021 at 1:12 PM Kenneth Knowles wrote: >>> Awesome! Just going to add a few colleagues (who are subscribed anyhow) to make sure this hits the top of their inbox. +Robert Burke +Chamikara Jayalath +Kyle Weaver Kenn On Thu, Nov 4, 2021 at 11:40 PM Jeff Rhyason wrote: > I'm interested to see the Go SDK work with the Spark runner. Based on > the instructions at https://beam.apache.org/get-started/quickstart-go/, > I run these commands and get the following failure: > > $ ./gradlew :runners:spark:2:job-server:runShadow > in another window: > $ cd sdks > $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo > --runner spark --endpoint localhost:8099 > 2021/11/04 22:06:46 No environment config specified. Using default > config: 'apache/beam_go_sdk:2.35.0.dev' > 2021/11/04 22:06:46 Failed to execute job: generating model > pipeline > failed to add scope tree: &{{CountWords root/CountWords} > [{main.extractFn 5: ParDo [In(Main): string <- {4: string/string GLO}] -> > [Out: string -> {5: string/string GLO}]}] [0xc96cd0]} > caused by: > failed to add input kind: {main.extractFn 5: ParDo [In(Main): string > <- {4: string/string GLO}] -> [Out: string -> {5: string/string GLO}]} > caused by: > failed to serialize 5: ParDo [In(Main): string <- {4: string/string > GLO}] -> [Out: string -> {5: string/string GLO}] > caused by: > encoding userfn 5: ParDo [In(Main): string <- {4: > string/string GLO}] -> [Out: string -> {5: string/string GLO}] > bad userfn > caused by: > encoding structural DoFn &{ 0xc000460ae8 > map[ProcessElement:0xc0004fcac0] map[]} > receiver type *main.extractFn must be registered > exit status 1 > > I was able to register that type, like this: > > diff --git a/sdks/go/examples/wordcount/wordcount.go > b/sdks/go/examples/wordcount/wordcount.go > index 4d54db9a2d..6db99d6220 100644 > --- a/sdks/go/examples/wordcount/wordcount.go > +++ b/sdks/go/examples/wordcount/wordcount.go > @@ -60,6 +60,7 @@ import ( > "flag" > "fmt" > "log" > + "reflect" > "regexp" > "strings" > > @@ -107,6 +108,7 @@ var ( > // by calling beam.RegisterFunction in an init() call. > func init() { > beam.RegisterFunction(formatFn) > + beam.RegisterDoFn(reflect.TypeOf((*extractFn)(nil)).Elem()) > } > > var ( > > > Then I encountered: > > $ go1.16.4 run ./go/examples/wordcount/wordcount.go --output foo > --runner spark --endpoint localhost:8099 > ... > 2021/11/04 23:07:26 (): java.lang.IllegalArgumentException: > Unsupported class file major version 55 > 2021/11/04 23:07:26 Job state: FAILED