forked from apache/beam
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[BEAM-14081] [CdapIO] Add context classes for CDAP plugins (apache#17104
) * [BEAM-14048] Add ConfigWrapper for building CDAP PluginConfigs * [BEAM-14048] Fix checkstyle * [BEAM-14048] Fix warnings * [BEAM-14048] Fix warnings * [BEAM-14048] Fix warning * [BEAM-14048] Fix warning * [BEAM-14048] Remove unused dependencies * [BEAM-14048] Add needed dependencies * [BEAM-14048] Fix spotless * [BEAM-14048] Fix typo * [BEAM-14048] Use fori instead of stream * [BEAM-14048] Suppress warning * [BEAM-14048] Add used undeclared artifacts * [BEAM-14048] Change dependencies to test * Add context. * Fix dependencies issue * Add null annotation * [BEAM-14048] Refactoring * Add SuppressWarning. * Fix style. * Determine dependencies. * [BEAM-14048] Use CDAP InstantiatorFactory for creating config objects * [BEAM-14048] Suppress warning * [BEAM-14081] Refactoring * Update maven repo * Update build.gradle * [BEAM-14081] Refactoring * [BEAM-14048] Use ServiceNow CDAP dependency from Maven central * [BEAM-14048] Set macroFields * [BEAM-14081] Fix javadoc * [BEAM-14081] Make BatchContextImpl class abstract Co-authored-by: vitaly.terentyev <vitaly.terentyev@akvelon.com> Co-authored-by: Alex Kosolapov <alex.kosolapov@gmail.com> Co-authored-by: Elizaveta Lomteva <elizaveta.lomteva@akvelon.com> Co-authored-by: Elizaveta Lomteva <57974525+Lizzfox@users.noreply.github.com>
- Loading branch information
5 people
committed
May 11, 2022
1 parent
43cc865
commit f30e4e5
Showing
10 changed files
with
585 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
232 changes: 232 additions & 0 deletions
232
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/context/BatchContextImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,232 @@ | ||
| /* | ||
| * 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.sdk.io.cdap.context; | ||
|
|
||
| import io.cdap.cdap.api.data.DatasetInstantiationException; | ||
| import io.cdap.cdap.api.data.batch.InputFormatProvider; | ||
| import io.cdap.cdap.api.data.schema.Schema; | ||
| import io.cdap.cdap.api.dataset.Dataset; | ||
| import io.cdap.cdap.api.dataset.DatasetManagementException; | ||
| import io.cdap.cdap.api.dataset.DatasetProperties; | ||
| import io.cdap.cdap.api.metadata.Metadata; | ||
| import io.cdap.cdap.api.metadata.MetadataEntity; | ||
| import io.cdap.cdap.api.metadata.MetadataException; | ||
| import io.cdap.cdap.api.metadata.MetadataScope; | ||
| import io.cdap.cdap.api.plugin.PluginProperties; | ||
| import io.cdap.cdap.etl.api.FailureCollector; | ||
| import io.cdap.cdap.etl.api.Lookup; | ||
| import io.cdap.cdap.etl.api.StageMetrics; | ||
| import io.cdap.cdap.etl.api.SubmitterLifecycle; | ||
| import io.cdap.cdap.etl.api.action.SettableArguments; | ||
| import io.cdap.cdap.etl.api.batch.BatchContext; | ||
| import io.cdap.cdap.etl.api.lineage.field.FieldOperation; | ||
| import java.net.URL; | ||
| import java.sql.Timestamp; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import javax.annotation.Nullable; | ||
|
|
||
| /** Class for Batch, Sink and Stream CDAP wrapper classes that use it to provide common details. */ | ||
| @SuppressWarnings({"TypeParameterUnusedInFormals", "nullness"}) | ||
| public abstract class BatchContextImpl implements BatchContext { | ||
|
|
||
| private final FailureCollectorWrapper failureCollector = new FailureCollectorWrapper(); | ||
|
|
||
| /** | ||
| * This should be set after {@link SubmitterLifecycle#prepareRun(Object)} call with passing this | ||
| * context object as a param. | ||
| */ | ||
| protected InputFormatProvider inputFormatProvider; | ||
|
|
||
| private final Timestamp startTime = new Timestamp(System.currentTimeMillis()); | ||
|
|
||
| public InputFormatProvider getInputFormatProvider() { | ||
| return inputFormatProvider; | ||
| } | ||
|
|
||
| @Override | ||
| public String getStageName() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public String getNamespace() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public String getPipelineName() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public long getLogicalStartTime() { | ||
| return this.startTime.getTime(); | ||
| } | ||
|
|
||
| @Override | ||
| public StageMetrics getMetrics() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public PluginProperties getPluginProperties() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public PluginProperties getPluginProperties(String pluginId) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public <T> Class<T> loadPluginClass(String pluginId) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public <T> T newPluginInstance(String pluginId) throws InstantiationException { | ||
| return null; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public Schema getInputSchema() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public @Nullable Map<String, Schema> getInputSchemas() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public @Nullable Schema getOutputSchema() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Map<String, Schema> getOutputPortSchemas() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public void createDataset(String datasetName, String typeName, DatasetProperties properties) | ||
| throws DatasetManagementException {} | ||
|
|
||
| @Override | ||
| public boolean datasetExists(String datasetName) throws DatasetManagementException { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public SettableArguments getArguments() { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public FailureCollector getFailureCollector() { | ||
| return this.failureCollector; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public URL getServiceURL(String applicationId, String serviceId) { | ||
| return null; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public URL getServiceURL(String serviceId) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Map<MetadataScope, Metadata> getMetadata(MetadataEntity metadataEntity) | ||
| throws MetadataException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Metadata getMetadata(MetadataScope scope, MetadataEntity metadataEntity) | ||
| throws MetadataException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public void addProperties(MetadataEntity metadataEntity, Map<String, String> properties) {} | ||
|
|
||
| @Override | ||
| public void addTags(MetadataEntity metadataEntity, String... tags) {} | ||
|
|
||
| @Override | ||
| public void addTags(MetadataEntity metadataEntity, Iterable<String> tags) {} | ||
|
|
||
| @Override | ||
| public void removeMetadata(MetadataEntity metadataEntity) {} | ||
|
|
||
| @Override | ||
| public void removeProperties(MetadataEntity metadataEntity) {} | ||
|
|
||
| @Override | ||
| public void removeProperties(MetadataEntity metadataEntity, String... keys) {} | ||
|
|
||
| @Override | ||
| public void removeTags(MetadataEntity metadataEntity) {} | ||
|
|
||
| @Override | ||
| public void removeTags(MetadataEntity metadataEntity, String... tags) {} | ||
|
|
||
| @Override | ||
| public void record(List<FieldOperation> fieldOperations) {} | ||
|
|
||
| @Override | ||
| public <T extends Dataset> T getDataset(String name) throws DatasetInstantiationException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public <T extends Dataset> T getDataset(String namespace, String name) | ||
| throws DatasetInstantiationException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public <T extends Dataset> T getDataset(String name, Map<String, String> arguments) | ||
| throws DatasetInstantiationException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public <T extends Dataset> T getDataset( | ||
| String namespace, String name, Map<String, String> arguments) | ||
| throws DatasetInstantiationException { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public void releaseDataset(Dataset dataset) {} | ||
|
|
||
| @Override | ||
| public void discardDataset(Dataset dataset) {} | ||
|
|
||
| @Override | ||
| public <T> Lookup<T> provide(String table, Map<String, String> arguments) { | ||
| return null; | ||
| } | ||
| } |
33 changes: 33 additions & 0 deletions
33
.../java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/context/BatchSinkContextImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * 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.sdk.io.cdap.context; | ||
|
|
||
| import io.cdap.cdap.api.data.batch.Output; | ||
| import io.cdap.cdap.etl.api.batch.BatchSinkContext; | ||
|
|
||
| /** Class for creating context object of different CDAP classes with batch sink type. */ | ||
| public class BatchSinkContextImpl extends BatchContextImpl implements BatchSinkContext { | ||
|
|
||
| @Override | ||
| public void addOutput(Output output) {} | ||
|
|
||
| @Override | ||
| public boolean isPreviewEnabled() { | ||
| return false; | ||
| } | ||
| } |
40 changes: 40 additions & 0 deletions
40
...ava/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/context/BatchSourceContextImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| /* | ||
| * 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.sdk.io.cdap.context; | ||
|
|
||
| import io.cdap.cdap.api.data.batch.Input; | ||
| import io.cdap.cdap.etl.api.batch.BatchSourceContext; | ||
|
|
||
| /** Class for creating context object of different CDAP classes with batch source type. */ | ||
| public class BatchSourceContextImpl extends BatchContextImpl implements BatchSourceContext { | ||
|
|
||
| @Override | ||
| public void setInput(Input input) { | ||
| this.inputFormatProvider = ((Input.InputFormatProviderInput) input).getInputFormatProvider(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isPreviewEnabled() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public int getMaxPreviewRecords() { | ||
| return 0; | ||
| } | ||
| } |
Oops, something went wrong.