Merge branch 'master' of https://github.com/apache/incubator-beam into apex-runner
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/320d6193 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/320d6193 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/320d6193 Branch: refs/heads/master Commit: 320d6193d8cd41aa99dce01aa8f8aabac30064fa Parents: 9900157 49ce270 Author: Thomas Weise <[email protected]> Authored: Fri Nov 11 21:33:38 2016 +0100 Committer: Thomas Weise <[email protected]> Committed: Fri Nov 11 21:33:38 2016 +0100 ---------------------------------------------------------------------- examples/java/pom.xml | 87 +- .../beam/examples/DebuggingWordCount.java | 67 +- .../apache/beam/examples/MinimalWordCount.java | 50 +- .../org/apache/beam/examples/WordCount.java | 62 +- .../beam/examples/DebuggingWordCountTest.java | 15 +- .../beam/examples/WindowedWordCountIT.java | 30 +- .../org/apache/beam/examples/WordCountIT.java | 17 +- .../examples/cookbook/BigQueryTornadoesIT.java | 16 +- examples/java8/pom.xml | 13 + .../beam/examples/complete/game/GameStats.java | 8 +- .../examples/complete/game/HourlyTeamScore.java | 8 +- .../examples/complete/game/LeaderBoard.java | 10 +- .../beam/examples/complete/game/UserScore.java | 6 +- .../examples/complete/game/GameStatsTest.java | 5 + .../complete/game/HourlyTeamScoreTest.java | 5 + .../examples/complete/game/LeaderBoardTest.java | 6 + examples/pom.xml | 5 + pom.xml | 34 + runners/apex/README.md | 2 +- .../beam/runners/core/SimpleDoFnRunner.java | 16 +- .../beam/runners/core/SplittableParDo.java | 25 +- .../core/UnboundedReadFromBoundedSource.java | 3 +- .../core/triggers/TriggerStateMachines.java | 16 +- runners/direct-java/pom.xml | 13 + .../beam/runners/direct/DirectRunner.java | 119 ++- .../runners/direct/DoFnLifecycleManager.java | 4 +- .../direct/EncodabilityEnforcementFactory.java | 80 -- .../ImmutabilityCheckingBundleFactory.java | 11 +- .../direct/ImmutabilityEnforcementFactory.java | 2 - .../direct/ImmutableListBundleFactory.java | 21 +- .../runners/direct/ParDoOverrideFactory.java | 2 +- .../direct/UnboundedReadEvaluatorFactory.java | 6 +- .../beam/runners/direct/WatermarkManager.java | 152 ++-- .../direct/CloningBundleFactoryTest.java | 122 ++- .../ConsumerTrackingPipelineVisitorTest.java | 17 + .../beam/runners/direct/DirectRunnerTest.java | 26 + .../EncodabilityEnforcementFactoryTest.java | 323 -------- .../direct/ImmutableListBundleFactoryTest.java | 15 +- .../UnboundedReadEvaluatorFactoryTest.java | 47 +- .../runners/direct/WatermarkManagerTest.java | 83 ++ .../beam/runners/flink/examples/TFIDF.java | 11 +- runners/flink/pom.xml | 1 + .../apache/beam/runners/flink/FlinkRunner.java | 4 +- .../FlinkStreamingTransformTranslators.java | 4 - .../SerializableFnAggregatorWrapper.java | 7 + .../wrappers/streaming/WindowDoFnOperator.java | 2 +- .../streaming/io/UnboundedSocketSource.java | 5 +- .../streaming/io/UnboundedSourceWrapper.java | 3 +- runners/google-cloud-dataflow-java/pom.xml | 28 +- .../dataflow/testing/TestDataflowRunner.java | 20 +- .../dataflow/util/DataflowTransport.java | 9 +- runners/pom.xml | 5 + runners/spark/pom.xml | 9 + .../beam/runners/spark/SparkContextOptions.java | 64 ++ .../runners/spark/SparkPipelineOptions.java | 36 +- .../spark/translation/SparkContextFactory.java | 19 +- .../SparkRunnerStreamingContextFactory.java | 3 +- .../runners/spark/ProvidedSparkContextTest.java | 6 +- .../streaming/KafkaStreamingTest.java | 4 +- sdks/java/core/pom.xml | 31 +- .../org/apache/beam/sdk/coders/AvroCoder.java | 126 +-- .../org/apache/beam/sdk/io/FileBasedSink.java | 134 +-- .../org/apache/beam/sdk/options/GcpOptions.java | 136 +-- .../beam/sdk/testing/BigqueryMatcher.java | 15 +- .../beam/sdk/testing/TestPipelineOptions.java | 6 + .../org/apache/beam/sdk/transforms/DoFn.java | 20 + .../beam/sdk/transforms/DoFnAdapters.java | 28 +- .../org/apache/beam/sdk/transforms/ParDo.java | 47 +- .../reflect/ByteBuddyDoFnInvokerFactory.java | 828 +++++++++++++++++++ .../reflect/ByteBuddyOnTimerInvokerFactory.java | 279 +++++++ .../transforms/reflect/DoFnInvokerFactory.java | 27 + .../sdk/transforms/reflect/DoFnInvokers.java | 658 +-------------- .../sdk/transforms/reflect/DoFnSignature.java | 67 +- .../sdk/transforms/reflect/DoFnSignatures.java | 67 +- .../sdk/transforms/reflect/OnTimerInvoker.java | 27 + .../reflect/OnTimerInvokerFactory.java | 36 + .../sdk/transforms/reflect/OnTimerInvokers.java | 42 + .../beam/sdk/transforms/windowing/WindowFn.java | 12 + .../apache/beam/sdk/util/CredentialFactory.java | 4 +- .../org/apache/beam/sdk/util/Credentials.java | 192 ----- .../beam/sdk/util/FileIOChannelFactory.java | 3 +- .../beam/sdk/util/GcpCredentialFactory.java | 41 +- .../beam/sdk/util/NoopCredentialFactory.java | 9 +- .../apache/beam/sdk/util/PubsubGrpcClient.java | 18 +- .../apache/beam/sdk/util/PubsubJsonClient.java | 9 +- .../apache/beam/sdk/util/TestCredential.java | 44 +- .../org/apache/beam/sdk/util/Transport.java | 9 +- .../apache/beam/sdk/coders/AvroCoderTest.java | 48 ++ .../apache/beam/sdk/io/FileBasedSinkTest.java | 28 +- .../org/apache/beam/sdk/io/XmlSinkTest.java | 4 +- .../apache/beam/sdk/testing/ExpectedLogs.java | 15 +- .../beam/sdk/testing/ExpectedLogsTest.java | 30 +- .../apache/beam/sdk/transforms/FlattenTest.java | 38 +- .../apache/beam/sdk/transforms/ParDoTest.java | 61 ++ .../transforms/reflect/DoFnInvokersTest.java | 124 +-- .../DoFnSignaturesProcessElementTest.java | 12 +- .../DoFnSignaturesSplittableDoFnTest.java | 39 +- .../transforms/reflect/DoFnSignaturesTest.java | 69 +- .../transforms/reflect/OnTimerInvokersTest.java | 109 +++ .../beam/sdk/util/PubsubGrpcClientTest.java | 9 +- sdks/java/extensions/sorter/README.md | 4 +- sdks/java/io/google-cloud-platform/pom.xml | 9 +- .../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 4 +- .../beam/sdk/io/gcp/bigtable/BigtableIO.java | 98 ++- .../sdk/io/gcp/bigtable/BigtableService.java | 6 + .../io/gcp/bigtable/BigtableServiceImpl.java | 5 + .../beam/sdk/io/gcp/datastore/DatastoreV1.java | 25 +- .../sdk/io/gcp/bigtable/BigtableIOTest.java | 83 +- .../sdk/io/gcp/bigtable/BigtableWriteIT.java | 15 +- .../beam/sdk/io/gcp/datastore/V1ReadIT.java | 15 +- .../sdk/io/gcp/datastore/V1TestOptions.java | 5 - .../beam/sdk/io/gcp/datastore/V1TestUtil.java | 35 +- .../beam/sdk/io/gcp/datastore/V1WriteIT.java | 9 +- .../org/apache/beam/sdk/io/jdbc/JdbcIOTest.java | 25 +- sdks/java/io/jms/pom.xml | 13 + sdks/java/io/kafka/pom.xml | 13 + sdks/java/io/kinesis/pom.xml | 8 +- .../beam/sdk/io/kinesis/CustomOptional.java | 21 +- .../beam/sdk/io/kinesis/KinesisRecord.java | 3 +- .../beam/sdk/io/kinesis/CustomOptionalTest.java | 10 + sdks/java/io/mongodb/pom.xml | 13 + .../src/main/java/DebuggingWordCount.java | 69 +- .../src/main/java/MinimalWordCount.java | 52 +- .../src/main/java/WindowedWordCount.java | 6 +- .../src/main/java/WordCount.java | 64 +- .../common/ExampleBigQueryTableOptions.java | 2 +- .../src/main/java/common/ExampleOptions.java | 5 + ...xamplePubsubTopicAndSubscriptionOptions.java | 2 +- .../java/common/ExamplePubsubTopicOptions.java | 2 +- .../src/main/java/common/ExampleUtils.java | 3 +- .../src/test/java/DebuggingWordCountTest.java | 15 +- .../src/test/java/WordCountTest.java | 7 +- .../update-examples-archetype.sh | 59 ++ sdks/java/microbenchmarks/pom.xml | 13 + .../transforms/DoFnInvokersBenchmark.java | 2 +- sdks/pom.xml | 5 + 136 files changed, 3663 insertions(+), 2352 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/320d6193/examples/java/pom.xml ---------------------------------------------------------------------- diff --cc examples/java/pom.xml index ac16bbd,f66fd36..01d1695 --- a/examples/java/pom.xml +++ b/examples/java/pom.xml @@@ -62,43 -53,38 +53,50 @@@ <dependency> <groupId>org.apache.beam</groupId> <artifactId>beam-runners-direct-java</artifactId> - <version>${project.version}</version> <scope>runtime</scope> - <optional>true</optional> </dependency> + </dependencies> + </profile> ++ <!-- Include the Apache Apex runner with -P apex-runner --> ++ <profile> ++ <id>apex-runner</id> ++ <dependencies> + <dependency> + <groupId>org.apache.beam</groupId> - <artifactId>beam-runners-flink_2.10</artifactId> - <version>${project.version}</version> ++ <artifactId>beam-runners-apex</artifactId> + <scope>runtime</scope> - <optional>true</optional> + </dependency> ++ </dependencies> ++ </profile> + + <!-- Include the Apache Flink runner with -P flink-runner --> + <profile> + <id>flink-runner</id> + <dependencies> <dependency> <groupId>org.apache.beam</groupId> - <artifactId>beam-runners-google-cloud-dataflow-java</artifactId> - <version>${project.version}</version> + <artifactId>beam-runners-flink_2.10</artifactId> <scope>runtime</scope> - <optional>true</optional> </dependency> + </dependencies> + </profile> + <!-- Include the Apache Spark runner -P spark-runner --> + <profile> + <id>spark-runner</id> + <dependencies> <dependency> <groupId>org.apache.beam</groupId> - <artifactId>beam-runners-apex</artifactId> - <version>${project.version}</version> + <artifactId>beam-runners-spark</artifactId> <scope>runtime</scope> - <optional>true</optional> </dependency> - <dependency> - <groupId>org.apache.beam</groupId> - <artifactId>beam-runners-spark</artifactId> - <version>${project.version}</version> + <groupId>org.apache.spark</groupId> + <artifactId>spark-streaming_2.10</artifactId> + <version>${spark.version}</version> <scope>runtime</scope> - <optional>true</optional> </dependency> - <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.10</artifactId> http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/320d6193/runners/apex/README.md ---------------------------------------------------------------------- diff --cc runners/apex/README.md index c9e47a1,0000000..7aa433e mode 100644,000000..100644 --- a/runners/apex/README.md +++ b/runners/apex/README.md @@@ -1,76 -1,0 +1,76 @@@ +<!-- + 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. +--> + +Apex Beam Runner ï´¾ApexâRunnerï´¿ +============================= + +ApexâRunner is a Runner for Apache Beam which executes Beam pipelines with Apache Apex as underlying engine. The runner has broad support for the Beam model and supports streaming and batch pipelines. + +[Apache Apex](http://apex.apache.org/) is a stream processing platform and framework for low-latency, high-throughput and fault-tolerant analytics applications on Apache Hadoop. Apex is Java based and also provides its own API for application development (native compositional and declarative Java API, SQL) with a comprehensive [operator library](https://github.com/apache/apex-malhar). Apex has a unified streaming architecture and can be used for real-time and batch processing. With its stateful stream processing architecture Apex can support all of the concepts in the Beam model (event time, triggers, watermarks etc.). + +##Status + +Apex-Runner is relatively new. It is fully functional and can currently be used to run pipelines in embedded mode. It does not take advantage of all the performance and scalability that Apex can deliver. This is expected to be addressed with upcoming work, leveraging features like incremental checkpointing, partitioning and operator affinity from Apex. Please see [JIRA](https://issues.apache.org/jira/issues/?jql=project%20%3D%20BEAM%20AND%20component%20%3D%20runner-apex%20AND%20resolution%20%3D%20Unresolved) and we welcome contributions! + +##Getting Started + +The following shows how to run the WordCount example that is provided with the source code on Apex (the example is identical with the one provided as part of the Beam examples). + +###Installing Beam + +To get the latest version of Beam with Apex-Runner, first clone the Beam repository: + +``` +git clone https://github.com/apache/incubatorâbeam +``` + +Then switch to the newly created directory and run Maven to build the Apache Beam: + +``` +cd incubatorâbeam +mvn clean install âDskipTests +``` + +Now Apache Beam and the Apex Runner are installed in your local Maven repository. + +###Running an Example + +Download something to count: + +``` +curl http://www.gutenberg.org/cache/epub/1128/pg1128.txt > /tmp/kinglear.txt +``` + +Run the pipeline, using the Apex runner: + +``` +cd examples/java - mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--inputFile=/tmp/kinglear.txt --output=/tmp/wordcounts.txt --runner=ApexRunner" -Pinclude-runners ++mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount -Dexec.args="--inputFile=/tmp/kinglear.txt --output=/tmp/wordcounts.txt --runner=ApexRunner" -Papex-runner +``` + +Once completed, there will be multiple output files with the base name given above: + +``` +$ ls /tmp/out-* +/tmp/out-00000-of-00003 /tmp/out-00001-of-00003 /tmp/out-00002-of-00003 +``` + +##Running pipelines on an Apex YARN cluster + +Coming soon. http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/320d6193/runners/pom.xml ----------------------------------------------------------------------
