You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ec...@apache.org on 2018/12/27 13:57:16 UTC
[beam] branch spark-runner_structured-streaming updated (44fd6c7 ->
017dcb9)
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a change to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git.
from 44fd6c7 Experiment over using spark Catalog to pass in Beam Source through spark Table
new 83f8487 Add source mocks
new 98d9049 fix mock, wire mock in translators and create a main test.
new e510711 Use raw WindowedValue so that spark Encoders could work (temporary)
new 017dcb9 clean deps
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
runners/spark-structured-streaming/build.gradle | 17 +--
.../translation/TranslationContext.java | 8 ++
.../translation/batch/PipelineTranslatorBatch.java | 2 +-
.../batch/ReadSourceTranslatorBatch.java | 3 +-
...tch.java => ReadSourceTranslatorMockBatch.java} | 48 +++------
.../translation/io/DatasetSourceMock.java | 114 +++++++++++++++++++++
.../spark/structuredstreaming/SourceTest.java | 16 +++
7 files changed, 159 insertions(+), 49 deletions(-)
copy runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/{ReadSourceTranslatorBatch.java => ReadSourceTranslatorMockBatch.java} (53%)
create mode 100644 runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
create mode 100644 runners/spark-structured-streaming/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SourceTest.java
[beam] 03/04: Use raw WindowedValue so that spark Encoders could
work (temporary)
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e510711f9e055193a18b1ddc8b4bca16b6d9428c
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Fri Dec 21 16:05:31 2018 +0100
Use raw WindowedValue so that spark Encoders could work (temporary)
---
.../translation/TranslationContext.java | 8 ++++++++
.../batch/ReadSourceTranslatorMockBatch.java | 20 +++++---------------
2 files changed, 13 insertions(+), 15 deletions(-)
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
index 0f2493d..fb36b37 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
@@ -115,6 +115,14 @@ public class TranslationContext {
}
}
+ //TODO: remove. It is just for testing
+ public void putDatasetRaw(PValue value, Dataset<WindowedValue> dataset) {
+ if (!datasets.containsKey(value)) {
+ datasets.put(value, dataset);
+ leaves.add(dataset);
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// PCollections methods
// --------------------------------------------------------------------------------------------
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
index 504a64d..4a509de 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
@@ -50,25 +50,15 @@ class ReadSourceTranslatorMockBatch<T>
Dataset<Row> rowDataset = dataStreamReader.load();
- MapFunction<Row, WindowedValue<Integer>> func = new MapFunction<Row, WindowedValue<Integer>>() {
- @Override public WindowedValue<Integer> call(Row value) throws Exception {
+ MapFunction<Row, WindowedValue> func = new MapFunction<Row, WindowedValue>() {
+ @Override public WindowedValue call(Row value) throws Exception {
//there is only one value put in each Row by the InputPartitionReader
- return value.<WindowedValue<Integer>>getAs(0);
+ return value.<WindowedValue>getAs(0);
}
};
- Dataset<WindowedValue<Integer>> dataset = rowDataset.map(func, new Encoder<WindowedValue<Integer>>() {
-
- @Override public StructType schema() {
- return null;
- }
-
- @Override public ClassTag<WindowedValue<Integer>> clsTag() {
- return scala.reflect.ClassTag$.MODULE$.<WindowedValue<Integer>>apply(WindowedValue.class);
- }
- });
+ Dataset<WindowedValue> dataset = rowDataset.map(func, Encoders.kryo(WindowedValue.class));
PCollection<T> output = (PCollection<T>) context.getOutput();
- context.putDataset(output, dataset);
- dataset.show();
+ context.putDatasetRaw(output, dataset);
}
}
[beam] 01/04: Add source mocks
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 83f84873649f42ffaf8c3e5a5c862b0df4262d9e
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Dec 20 15:38:11 2018 +0100
Add source mocks
---
.../batch/ReadSourceTranslatorMockBatch.java | 81 +++++++++++++++
.../translation/io/DatasetSourceMock.java | 114 +++++++++++++++++++++
2 files changed, 195 insertions(+)
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
new file mode 100644
index 0000000..5b1bada
--- /dev/null
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
@@ -0,0 +1,81 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
+
+import java.io.IOException;
+import org.apache.beam.runners.core.construction.ReadTranslation;
+import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
+import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
+import org.apache.beam.runners.spark.structuredstreaming.translation.io.DatasetSource;
+import org.apache.beam.runners.spark.structuredstreaming.translation.io.DatasetSourceMock;
+import org.apache.beam.sdk.io.BoundedSource;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoder;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.streaming.DataStreamReader;
+import org.apache.spark.sql.types.StructType;
+import scala.reflect.ClassTag;
+
+class ReadSourceTranslatorMockBatch<T>
+ implements TransformTranslator<PTransform<PBegin, PCollection<T>>> {
+
+ private String SOURCE_PROVIDER_CLASS = DatasetSourceMock.class.getCanonicalName();
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void translateTransform(
+ PTransform<PBegin, PCollection<T>> transform, TranslationContext context) {
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> rootTransform =
+ (AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>>)
+ context.getCurrentTransform();
+
+ String providerClassName = SOURCE_PROVIDER_CLASS.substring(0, SOURCE_PROVIDER_CLASS.indexOf("$"));
+ SparkSession sparkSession = context.getSparkSession();
+ DataStreamReader dataStreamReader = sparkSession.readStream().format(providerClassName);
+
+ Dataset<Row> rowDataset = dataStreamReader.load();
+
+ MapFunction<Row, WindowedValue<Integer>> func = new MapFunction<Row, WindowedValue<Integer>>() {
+ @Override public WindowedValue<Integer> call(Row value) throws Exception {
+ //there is only one value put in each Row by the InputPartitionReader
+ return value.<WindowedValue<Integer>>getAs(0);
+ }
+ };
+ Dataset<WindowedValue<Integer>> dataset = rowDataset.map(func, new Encoder<WindowedValue<Integer>>() {
+
+ @Override public StructType schema() {
+ return null;
+ }
+
+ @Override public ClassTag<WindowedValue<Integer>> clsTag() {
+ return scala.reflect.ClassTag$.MODULE$.<WindowedValue<Integer>>apply(WindowedValue.class);
+ }
+ });
+
+ PCollection<T> output = (PCollection<T>) context.getOutput();
+ context.putDataset(output, dataset);
+ }
+}
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
new file mode 100644
index 0000000..fa42fdf
--- /dev/null
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
@@ -0,0 +1,114 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.spark.structuredstreaming.translation.io;
+
+import static scala.collection.JavaConversions.asScalaBuffer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.apache.beam.sdk.io.BoundedSource;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.sources.v2.DataSourceOptions;
+import org.apache.spark.sql.sources.v2.DataSourceV2;
+import org.apache.spark.sql.sources.v2.MicroBatchReadSupport;
+import org.apache.spark.sql.sources.v2.reader.InputPartition;
+import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
+import org.apache.spark.sql.sources.v2.reader.streaming.MicroBatchReader;
+import org.apache.spark.sql.sources.v2.reader.streaming.Offset;
+import org.apache.spark.sql.types.StructType;
+import org.joda.time.Instant;
+
+/**
+ * This is a mock source that gives values between 0 and 999.
+ */
+public class DatasetSourceMock implements DataSourceV2, MicroBatchReadSupport {
+
+ @Override public MicroBatchReader createMicroBatchReader(Optional<StructType> schema, String checkpointLocation, DataSourceOptions options) {
+ return new DatasetMicroBatchReader();
+ }
+
+ /** This class can be mapped to Beam {@link BoundedSource}. */
+ private class DatasetMicroBatchReader implements MicroBatchReader {
+
+ @Override public void setOffsetRange(Optional<Offset> start, Optional<Offset> end) {
+ }
+
+ @Override public Offset getStartOffset() {
+ return null;
+ }
+
+ @Override public Offset getEndOffset() {
+ return null;
+ }
+
+ @Override public Offset deserializeOffset(String json) {
+ return null;
+ }
+
+ @Override public void commit(Offset end) {
+ }
+
+ @Override public void stop() {
+ }
+
+ @Override public StructType readSchema() {
+ return null;
+ }
+
+ @Override public List<InputPartition<InternalRow>> planInputPartitions() {
+ List<InputPartition<InternalRow>> result = new ArrayList<>();
+ result.add(new InputPartition<InternalRow>() {
+
+ @Override public InputPartitionReader<InternalRow> createPartitionReader() {
+ return new DatasetMicroBatchPartitionReaderMock();
+ }
+ });
+ return result;
+ }
+ }
+
+ /** This class is a mocked reader*/
+ private class DatasetMicroBatchPartitionReaderMock implements InputPartitionReader<InternalRow> {
+
+ private ArrayList<Integer> values;
+ private int currentIndex = 0;
+
+ private DatasetMicroBatchPartitionReaderMock() {
+ for (int i = 0; i < 1000; i++){
+ values.add(i);
+ }
+ }
+
+ @Override public boolean next() throws IOException {
+ currentIndex++;
+ return (currentIndex <= values.size());
+ }
+
+ @Override public void close() throws IOException {
+ }
+
+ @Override public InternalRow get() {
+ List<Object> list = new ArrayList<>();
+ list.add(WindowedValue.timestampedValueInGlobalWindow(values.get(currentIndex), new Instant()));
+ return InternalRow.apply(asScalaBuffer(list).toList());
+ }
+ }
+}
\ No newline at end of file
[beam] 02/04: fix mock,
wire mock in translators and create a main test.
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 98d9049560a40ff5245b6bcc972cb6424fd318a9
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Dec 20 17:18:54 2018 +0100
fix mock, wire mock in translators and create a main test.
---
.../translation/batch/PipelineTranslatorBatch.java | 2 +-
.../batch/ReadSourceTranslatorBatch.java | 3 ++-
.../batch/ReadSourceTranslatorMockBatch.java | 21 +++++++--------------
.../translation/io/DatasetSourceMock.java | 6 +++---
.../spark/structuredstreaming/SourceTest.java | 16 ++++++++++++++++
5 files changed, 29 insertions(+), 19 deletions(-)
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
index 26f1b9c..9ccc712 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
@@ -65,7 +65,7 @@ public class PipelineTranslatorBatch extends PipelineTranslator {
PTransformTranslation.PAR_DO_TRANSFORM_URN, new ParDoTranslatorBatch());
TRANSFORM_TRANSLATORS.put(
- PTransformTranslation.READ_TRANSFORM_URN, new ReadSourceTranslatorBatch());
+ PTransformTranslation.READ_TRANSFORM_URN, new ReadSourceTranslatorMockBatch());
}
public PipelineTranslatorBatch(SparkPipelineOptions options) {
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorBatch.java
index 0b828fb..aed016a 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorBatch.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorBatch.java
@@ -40,6 +40,7 @@ import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalog.Catalog;
import org.apache.spark.sql.catalyst.catalog.CatalogTable;
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
import org.apache.spark.sql.streaming.DataStreamReader;
class ReadSourceTranslatorBatch<T>
@@ -63,8 +64,8 @@ class ReadSourceTranslatorBatch<T>
throw new RuntimeException(e);
}
SparkSession sparkSession = context.getSparkSession();
-
DataStreamReader dataStreamReader = sparkSession.readStream().format(providerClassName);
+
Dataset<Row> rowDataset = dataStreamReader.load();
//TODO initialize source : how, to get a reference to the DatasetSource instance that spark
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
index 5b1bada..504a64d 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ReadSourceTranslatorMockBatch.java
@@ -17,28 +17,25 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import java.io.IOException;
-import org.apache.beam.runners.core.construction.ReadTranslation;
import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.io.DatasetSource;
import org.apache.beam.runners.spark.structuredstreaming.translation.io.DatasetSourceMock;
-import org.apache.beam.sdk.io.BoundedSource;
-import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.DataStreamReader;
-import org.apache.spark.sql.types.StructType;
-import scala.reflect.ClassTag;
+
+/**
+ * Mock translator that generates a source of 0 to 999 and prints it.
+ * @param <T>
+ */
class ReadSourceTranslatorMockBatch<T>
implements TransformTranslator<PTransform<PBegin, PCollection<T>>> {
@@ -48,13 +45,8 @@ class ReadSourceTranslatorMockBatch<T>
@Override
public void translateTransform(
PTransform<PBegin, PCollection<T>> transform, TranslationContext context) {
- AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> rootTransform =
- (AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>>)
- context.getCurrentTransform();
-
- String providerClassName = SOURCE_PROVIDER_CLASS.substring(0, SOURCE_PROVIDER_CLASS.indexOf("$"));
SparkSession sparkSession = context.getSparkSession();
- DataStreamReader dataStreamReader = sparkSession.readStream().format(providerClassName);
+ DataStreamReader dataStreamReader = sparkSession.readStream().format(SOURCE_PROVIDER_CLASS);
Dataset<Row> rowDataset = dataStreamReader.load();
@@ -77,5 +69,6 @@ class ReadSourceTranslatorMockBatch<T>
PCollection<T> output = (PCollection<T>) context.getOutput();
context.putDataset(output, dataset);
+ dataset.show();
}
}
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
index fa42fdf..ec88364 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/io/DatasetSourceMock.java
@@ -46,7 +46,7 @@ public class DatasetSourceMock implements DataSourceV2, MicroBatchReadSupport {
}
/** This class can be mapped to Beam {@link BoundedSource}. */
- private class DatasetMicroBatchReader implements MicroBatchReader {
+ private static class DatasetMicroBatchReader implements MicroBatchReader {
@Override public void setOffsetRange(Optional<Offset> start, Optional<Offset> end) {
}
@@ -70,7 +70,7 @@ public class DatasetSourceMock implements DataSourceV2, MicroBatchReadSupport {
}
@Override public StructType readSchema() {
- return null;
+ return new StructType();
}
@Override public List<InputPartition<InternalRow>> planInputPartitions() {
@@ -86,7 +86,7 @@ public class DatasetSourceMock implements DataSourceV2, MicroBatchReadSupport {
}
/** This class is a mocked reader*/
- private class DatasetMicroBatchPartitionReaderMock implements InputPartitionReader<InternalRow> {
+ private static class DatasetMicroBatchPartitionReaderMock implements InputPartitionReader<InternalRow> {
private ArrayList<Integer> values;
private int currentIndex = 0;
diff --git a/runners/spark-structured-streaming/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SourceTest.java b/runners/spark-structured-streaming/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SourceTest.java
new file mode 100644
index 0000000..eea9769
--- /dev/null
+++ b/runners/spark-structured-streaming/src/test/java/org/apache/beam/runners/spark/structuredstreaming/SourceTest.java
@@ -0,0 +1,16 @@
+package org.apache.beam.runners.spark.structuredstreaming;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Create;
+
+public class SourceTest {
+ public static void main(String[] args) {
+ PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
+ Pipeline pipeline = Pipeline.create(options);
+ pipeline.apply(Create.of(1));
+ pipeline.run();
+ }
+
+}
[beam] 04/04: clean deps
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 017dcb932964ce779ee6f789d2b66a55bcb03c1b
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Fri Dec 21 16:24:23 2018 +0100
clean deps
---
runners/spark-structured-streaming/build.gradle | 17 ++++-------------
1 file changed, 4 insertions(+), 13 deletions(-)
diff --git a/runners/spark-structured-streaming/build.gradle b/runners/spark-structured-streaming/build.gradle
index b33a2b6..803058f 100644
--- a/runners/spark-structured-streaming/build.gradle
+++ b/runners/spark-structured-streaming/build.gradle
@@ -45,36 +45,27 @@ test {
}
dependencies {
+ compile('org.apache.spark:spark-core_2.11:2.4.0'){
+ exclude group: 'org.json4s', module: 'json4s-jackson_2.11'
+ }
+ shadow "org.json4s:json4s-jackson_2.11:3.6.3"
shadow project(path: ":beam-model-pipeline", configuration: "shadow")
shadow project(path: ":beam-sdks-java-core", configuration: "shadow")
shadow project(path: ":beam-runners-core-construction-java", configuration: "shadow")
shadow project(path: ":beam-runners-core-java", configuration: "shadow")
shadow library.java.guava
- shadow library.java.jackson_annotations
shadow library.java.slf4j_api
shadow library.java.joda_time
- shadow "io.dropwizard.metrics:metrics-core:3.1.2"
- shadow library.java.jackson_module_scala
provided library.java.spark_sql
- provided library.java.hadoop_common
- provided library.java.hadoop_mapreduce_client_core
provided library.java.commons_compress
provided library.java.commons_lang3
provided library.java.commons_io_2x
provided library.java.hamcrest_core
provided library.java.hamcrest_library
- provided "org.apache.zookeeper:zookeeper:3.4.11"
- provided "org.scala-lang:scala-library:2.11.8"
- provided "com.esotericsoftware.kryo:kryo:2.21"
- shadowTest project(path: ":beam-sdks-java-io-kafka", configuration: "shadow")
shadowTest project(path: ":beam-sdks-java-core", configuration: "shadowTest")
shadowTest project(path: ":beam-runners-core-java", configuration: "shadowTest")
- shadowTest library.java.avro
- shadowTest library.java.kafka_clients
shadowTest library.java.junit
shadowTest library.java.mockito_core
- shadowTest library.java.jackson_dataformat_yaml
- shadowTest "org.apache.kafka:kafka_2.11:0.11.0.1"
validatesRunner project(path: ":beam-sdks-java-core", configuration: "shadowTest")
validatesRunner project(path: project.path, configuration: "shadowTest")
validatesRunner project(path: project.path, configuration: "shadow")