You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "byronellis (via GitHub)" <gi...@apache.org> on 2023/08/18 14:52:14 UTC
[GitHub] [beam] byronellis opened a new pull request, #28062: Experimental Swift SDK Implementation
byronellis opened a new pull request, #28062:
URL: https://github.com/apache/beam/pull/28062
An experimental implementation of a Beam SDK for the Swift language. This implementation relies on the forthcoming Swift 5.9 compiler version and has been tested against the 8/11 beta release on Intel Mac and Linux platforms.
It provides basic support for pipeline construction and submission to a portable runner. An implementation of ParDo as well as most of the basic Coders. The IntegrationTests.swift file contains an example of word count which processes local files (included as test fixtures).
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] 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://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] byronellis commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306741820
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Interesting. The worker code is connecting to Prism on the port specified so that doesn't appear to be a problem, though it's a bit hard due to the "backwards" nature of the control plane connection. And, yes, the Swift toolchain is surprisingly pleasant to use. With the plugin system we should also be able to add some nice quality-of-life Beam features to SPM as well.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306705926
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
I've successfully installed Swift (5.9 dev snap) on my personal laptop and pulled down the PR locally, and can replicate the behaviour on prism that you were seeing. I should be able to get to the bottom of those problems, and make prism more robust!
Hot dang do I love the clarity in the Swift Docs on installing on different OS flavours, and how automated `swift test` is. It failed the first time, and clearly said what else I needed to install (openSSL) to clear it up.
We should also add the following to the beam repo's .gitignore, right under the python section.
```
# Ignore files generated by the Swift build process
sdks/swift/.build/**/*
```
Building the swift SDK lead to many many files in there.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306464350
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
I think where prism is failing with Swift is that I've currently got the code to be a bit blocking on state & message transitions, and I'm not sure I have it set up to "do the right thing" with the unary getState calls, vs getStateStream or getMessageStream, so eventually it gets blocked up.
Not a 100% explanation for what we've seen, but better than nothing.
Great info! I'll let you know when it's resolved.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] byronellis commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306489436
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
I tested on Linux x86 but haven't tried Linux ARM yet. In theory when I was messing around with timestamps I'm doing the right thing with endianness there but I haven't verified that everywhere yet. In any case, "swift test" should work with a beta 5.9 build of Swift. I'm currently moving some of that to an "Examples" folder to make this sort of thing easier. I mostly used tests initially because it's a one button command in my IDE.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306712392
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
The mystery problem has gone deeper. Prism is hosting worker services at X, swift is trying to connect to X, and then prism apparently never gets that connection.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] byronellis commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1307607562
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
I'm guessing it's case 3: A transform that is fused by the other runners that Prism doesn't fuse. From the error I'm guessing that Prism is asking for a length prefixed coder whereas the Swift SDK coder is not length prefixed. Swift's coder implementation is pretty strict right now and does not automatically convert non-optional coders into optional coders, but I think that should be allowable.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Experimental Swift SDK Implementation [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28062:
URL: https://github.com/apache/beam/pull/28062#issuecomment-1837139189
This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Experimental Swift SDK Implementation [beam]
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis closed pull request #28062: Experimental Swift SDK Implementation
URL: https://github.com/apache/beam/pull/28062
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306452602
##########
sdks/swift/Sources/ApacheBeam/Runtime/Bundle/Source.swift:
##########
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+import Logging
+
+/// Custom SerializableFn that reads/writes from an external data stream using a defined coder. It assumes that a given
+/// data element might contain more than one coder
+final class Source : SerializableFn {
+
+ let client: DataplaneClient
+ let coder: Coder
+ let log: Logger
+
+ public init(client: DataplaneClient,coder:Coder) {
+ self.client = client
+ self.coder = coder
+ self.log = Logger(label:"Source")
+ }
+
+
+ func process(context: SerializableFnBundleContext,
+ inputs: [AnyPCollectionStream], outputs: [AnyPCollectionStream]) async throws -> (String, String) {
+ log.info("Waiting for input on \(context.instruction)-\(context.transform)")
+ let (stream,_) = await client.makeStream(instruction: context.instruction, transform: context.transform)
+
+ var messages: Int = 0
+ var count: Int = 0
+ for await message in stream {
+ messages += 1
+ switch message {
+ case let .data(data):
+ var d = data
+ while d.count > 0 {
+ let value = try coder.decode(&d)
+ for output in outputs {
+ try output.emit(value: value)
+ count += 1
+ }
+ }
+ case let .last(id, transform):
+ for output in outputs {
+ output.finish()
+ }
+ await client.finalizeStream(instruction: id, transform: transform)
+ log.info("Source \(context.instruction),\(context.transform) handled \(count) items over \(messages) messages")
+ return (id,transform)
+ //TODO: Handle timer messages
+ default:
+ log.info("Unhanled message \(message)")
Review Comment:
```suggestion
log.info("Unhandled message \(message)")
```
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306874983
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Note for tomorrow: the tests don't fail if the pipeline fails.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306869495
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Not sure what's going on. Prism does the same thing with both the jobmanagement port and the worker port. I added a debugz page with thread dumps and it just shows prism is waiting on a worker to connect, so it can produce work.
In the short term I'll try and get prism to gracefully fail the job if it doesn't get a control connection in a minute.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Experimental Swift SDK Implementation [beam]
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis commented on PR #28062:
URL: https://github.com/apache/beam/pull/28062#issuecomment-1837756974
This PR has been obsoleted by beam-swift repo
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] byronellis commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "byronellis (via GitHub)" <gi...@apache.org>.
byronellis commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1307589016
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Ah, excellent. Is the problem that Go is using IPv6 and I'm connecting on IPv4? Most of the code I looked at seemed to be doing the localhost thing so I just sort of copied that... but that's probably suboptimal in the year 2023 (At some point I also need to add the TLS support too)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1307596643
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Here's the sequence of events:
I ask for a port ":0", to listen on on any IP this machine is on.
It gives me "[::]:PORT"
but [::] is ipv6 unspecified.
In Go, that's explicitly handed as the loopback address.
The fix on my end is to just say the endpoint is one of "localhost:PORT" or "[::1]:PORT".
The next failure does happen on Swift side (KV<String, String> isn't KV<String,Optional<String>>) which either means prism messed up coders, or the swift SDK did, and this is just revealing something due to prism executing a single transform.
I need to make prism gracefully mark the job failed on that SDK worker failure though.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306469707
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Back to stumped after replicated this code. Prism is happy to run this approach from the Go SDK side. I might try seeing how hard it is to get the Swift SDK running in Linux, and answer my questions directly.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1306906128
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
Got it!
I just let Go's net listener determine the endpoint host spec. For local host, it uses [::]:PORT, which it's fine with parsing itself when it receives it.
The bit that made it tricky is I didn't notice the GRPCChannel override to support the beam api endpoint descriptor. But when I did, i hard changed the host to "localhost" and then the pipeline fails with a reasonable "whats's a split request?" error.
Prisms at fault here, and has become more resilient as a result. #winning
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1307720740
##########
sdks/swift/Sources/ApacheBeam/Runners/PortableRunner.swift:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Logging
+import GRPC
+import NIOCore
+
+public struct PortableRunner : PipelineRunner {
+ let loopback:Bool
+ let log: Logging.Logger
+ let host: String
+ let port: Int
+
+ public init(host:String="localhost",port:Int=8073,loopback:Bool=false) {
+ self.loopback = loopback
+ self.log = .init(label: "PortableRunner")
+ self.host = host
+ self.port = port
+ }
+
+ public func run(_ context: PipelineContext) async throws {
+ var proto = context.proto
+ if loopback {
+ //If we are in loopback mode we want to replace the default environment
+ //with an external environment that points to our local worker server
+ let worker = try WorkerServer(context.collections,context.pardoFns)
+ log.info("Running in LOOPBACK mode with a worker server at \(worker.endpoint).")
+ proto.components.environments[context.defaultEnvironmentId] = try .with {
+ try Environment.external(worker.endpoint).populate(&$0)
+ }
+ }
+ log.info("\(proto)")
+ log.info("Connecting to Portable Runner at \(host):\(port).")
+ let group = PlatformSupport.makeEventLoopGroup(loopCount: 1)
+ let channel = try GRPCChannelPool.with(target: .host(host, port: port),
+ transportSecurity: .plaintext, eventLoopGroup: group)
+ let client = Org_Apache_Beam_Model_JobManagement_V1_JobServiceAsyncClient(channel: channel)
+ let prepared = try await client.prepare(.with {
+ $0.pipeline = proto
+ })
+ let job = try await client.run(.with {
+ $0.preparationID = prepared.preparationID
+ })
+ log.info("Submitted job \(job.jobID)")
+ var done = false
+ while !done {
+ let status = try await client.getState(.with {
Review Comment:
```
2023-08-28T09:35:31-0700 info Source : [ApacheBeam] Waiting for input on inst003-stage004_source
Could not cast value of type 'ApacheBeam.KV<Swift.String, Swift.Optional<Swift.String>>' (0x7f3a980093f8) to 'ApacheBeam.KV<Swift.String, Swift.String>' (0x7f3adf6cf438).
```
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #28062: Experimental Swift SDK Implementation
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #28062:
URL: https://github.com/apache/beam/pull/28062#issuecomment-1684061100
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/28062?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#28062](https://app.codecov.io/gh/apache/beam/pull/28062?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (7cfc6a7) into [master](https://app.codecov.io/gh/apache/beam/commit/767a2d72510bf0d6d100e80f45c74ede913ece30?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (767a2d7) will **decrease** coverage by `0.01%`.
> Report is 32 commits behind head on master.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #28062 +/- ##
==========================================
- Coverage 72.33% 72.32% -0.01%
==========================================
Files 678 678
Lines 99726 99730 +4
==========================================
- Hits 72133 72132 -1
- Misses 26030 26035 +5
Partials 1563 1563
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/28062/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/28062/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.86% <ø> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 12 files with indirect coverage changes](https://app.codecov.io/gh/apache/beam/pull/28062/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] liferoad commented on a diff in pull request #28062: Experimental Swift SDK Implementation
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #28062:
URL: https://github.com/apache/beam/pull/28062#discussion_r1298863247
##########
sdks/python/output.json-00000-of-00001:
##########
@@ -0,0 +1,2 @@
+{"col1":"bar","col2":2,"col3":200}
Review Comment:
Remove this?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org