Add some auto-starting runners to the typescript SDK.

sdks/java/extensions/python/src/main/resources/org/apache/beam/sdk/extensions/python/bootstrap_beam_venv.py

@@ -77,7 +77,7 @@ def maybe_strict_version(s):
  or options.beam_version.startswith('https://')):
  # It's a path to a tarball.
  beam_version = os.path.basename(options.beam_version)
- beam_package = options.beam_version
+ beam_package = options.beam_version + '[gcp,aws,asure,dataframe]'
  else:
  beam_version = options.beam_version
  beam_package = 'apache_beam[gcp,aws,asure,dataframe]==' + beam_version

sdks/typescript/package.json

@@ -32,6 +32,7 @@
  "chai": "^4.3.4",
  "date-fns": "^2.28.0",
  "fast-deep-equal": "^3.1.3",
+ "find-git-root": "^1.0.4",
  "long": "^4.0.0",
  "protobufjs": "^6.10.2",
  "queue-typescript": "^1.0.1",

sdks/typescript/src/apache_beam/examples/wordcount.ts

@@ -16,10 +16,24 @@
  * limitations under the License.
  */
 
-// TODO: Should this be in a top-level examples dir, rather than under apache_beam.
+// Run directly with
+//
+// node dist/src/apache_beam/examples/wordcount.js
+//
+// A different runner can be chosen via a --runner argument, e.g.
+//
+// node dist/src/apache_beam/examples/wordcount.js --runner=flink
+//
+// To run on Dataflow, pass the required arguments:
+//
+// node dist/src/apache_beam/examples/wordcount.js --runner=dataflow --project=PROJECT_ID --tempLocation=gs://BUCKET/DIR' --region=us-central1
+
+// TODO: Should this be in a top-level examples dir, rather than under apache_beam?
+
+import * as yargs from "yargs";
 
 import * as beam from "../../apache_beam";
-import { DirectRunner } from "../runners/direct_runner";
+import { createRunner } from "../runners/runner";
 
 import { count } from "../transforms/combiners";
 import { GroupBy } from "../transforms/group_and_combine";
@@ -45,7 +59,7 @@ function wordCount(lines: beam.PCollection<string>): beam.PCollection<any> {
 }
 
 async function main() {
- await new DirectRunner().run((root) => {
+ await createRunner(yargs.argv).run((root) => {
  const lines = root.apply(
  new beam.Create([
  "In the beginning God created the heaven and the earth.",

sdks/typescript/src/apache_beam/runners/dataflow.ts

@@ -0,0 +1,44 @@
+/*
+ * 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 { Pipeline } from "../internal/pipeline";
+import { PipelineResult, Runner } from "./runner";
+import { PortableRunner } from "./portable_runner/runner";
+import { PythonService } from "../utils/service";
+
+export function dataflowRunner(runnerOptions: {
+ project: string;
+ tempLocation: string;
+ region: string;
+ [others: string]: any;
+}): Runner {
+ return new (class extends Runner {
+ async runPipeline(
+ pipeline: Pipeline,
+ options: Object = {}
+ ): Promise<PipelineResult> {
+ return new PortableRunner(
+ runnerOptions as any,
+ new PythonService("apache_beam.runners.dataflow.dataflow_job_service", [
+ "--port",
+ "{{PORT}}",
+ ])
+ ).runPipeline(pipeline, options);
+ }
+ })();
+}

sdks/typescript/src/apache_beam/runners/direct_runner.ts

@@ -44,14 +44,61 @@ import {
 } from "../values";
 import { PaneInfoCoder } from "../coders/standard_coders";
 import { Coder, Context as CoderContext } from "../coders/coders";
+import * as environments from "../internal/environments";
 import { serializeFn, deserializeFn } from "../internal/serialize";
 
+const SUPPORTED_REQUIREMENTS: string[] = [];
+
+export function directRunner(options: Object = {}): Runner {
+ return new DirectRunner(options);
+}
+
 export class DirectRunner extends Runner {
  // All the operators for a given pipeline should share the same state.
  // This global mapping allows operators to look up a shared state object for
  // a given pipeline on deserialization.
  static inMemoryStatesRefs: Map<string, InMemoryStateProvider> = new Map();
 
+ constructor(private options: Object = {}) {
+ super();
+ }
+
+ unsupportedFeatures(pipeline, options: Object = {}): string[] {
+ return [...this.unsupportedFeaturesIter(pipeline, options)];
+ }
+
+ *unsupportedFeaturesIter(pipeline, options: Object = {}) {
+ const proto: runnerApi.Pipeline = pipeline.proto;
+ for (const requirement of proto.requirements) {
+ if (!SUPPORTED_REQUIREMENTS.includes(requirement)) {
+ yield requirement;
+ }
+ }
+
+ for (const env of Object.values(proto.components!.environments)) {
+ if (
+ env.urn &&
+ env.urn != environments.TYPESCRIPT_DEFAULT_ENVIRONMENT_URN
+ ) {
+ yield env.urn;
+ }
+ }
+
+ for (const windowing of Object.values(
+ proto.components!.windowingStrategies
+ )) {
+ if (
+ ![
+ runnerApi.MergeStatus_Enum.UNSPECIFIED,
+ runnerApi.MergeStatus_Enum.NON_MERGING,
+ runnerApi.MergeStatus_Enum.ALREADY_MERGED,
+ ].includes(windowing.mergeStatus)
+ ) {
+ yield "MergeStatus=" + windowing.mergeStatus;
+ }
+ }
+ }
+
  async runPipeline(p): Promise<PipelineResult> {
  // console.dir(p.proto, { depth: null });
 

sdks/typescript/src/apache_beam/runners/flink.ts

@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+const fs = require("fs");
+const os = require("os");
+const path = require("path");
+
+import { Pipeline } from "../internal/pipeline";
+import { PipelineResult, Runner } from "./runner";
+import { PortableRunner } from "./portable_runner/runner";
+import { JavaJarService } from "../utils/service";
+
+const MAGIC_HOST_NAMES = ["[local]", "[auto]"];
+
+// These should stay in sync with gradle.properties.
+const PUBLISHED_FLINK_VERSIONS = ["1.12", "1.13", "1.14"];
+
+const defaultOptions = {
+ flinkMaster: "[local]",
+ flinkVersion: PUBLISHED_FLINK_VERSIONS[PUBLISHED_FLINK_VERSIONS.length - 1],
+};
+
+export function flinkRunner(runnerOptions: Object = {}): Runner {
+ return new (class extends Runner {
+ async runPipeline(
+ pipeline: Pipeline,
+ options: Object = {}
+ ): Promise<PipelineResult> {
+ const allOptions = {
+ ...defaultOptions,
+ ...runnerOptions,
+ ...options,
+ } as any;
+ if (
+ !allOptions.environmentType &&
+ MAGIC_HOST_NAMES.includes(allOptions.flinkMaster)
+ ) {
+ allOptions.environmentType = "LOOPBACK";
+ }
+ if (!allOptions.artifactsDir) {
+ allOptions.artifactsDir = fs.mkdtempSync(
+ path.join(os.tmpdir(), "flinkArtifactsDir")
+ );
+ }
+
+ const jobServerJar =
+ allOptions.flinkJobServerJar ||
+ (await JavaJarService.cachedJar(
+ JavaJarService.gradleToJar(
+ `runners:flink:${allOptions.flinkVersion}:job-server:shadowJar`
+ )
+ ));
+ const jobServer = new JavaJarService(jobServerJar, [
+ "--flink-master",
+ allOptions.flinkMaster,
+ "--artifacts-dir",
+ allOptions.artifactsDir,
+ "--job-port",
+ "{{PORT}}",
+ "--artifact-port",
+ "0",
+ "--expansion-port",
+ "0",
+ ]);
+
+ return new PortableRunner(allOptions, jobServer).runPipeline(pipeline);
+ }
+ })();
+}