You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/08/04 11:32:56 UTC
[GitHub] [beam] Amar3tto opened a new pull request, #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Amar3tto opened a new pull request, #22584:
URL: https://github.com/apache/beam/pull/22584
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] 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://beam.apache.org/contribute/get-started-contributing/#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.
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r938509466
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
+ try {
+ Constructor<?> constructor =
+ getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
+ constructor.setAccessible(true);
+ cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
+ } catch (Exception e) {
+ LOG.error("Can not instantiate CDAP plugin class", e);
+ throw new IllegalStateException("Can not call prepareRun");
}
- } else {
- for (Map.Entry<String, String> entry :
- getContext().getOutputFormatProvider().getOutputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ }
+ try {
+ cdapPluginObj.prepareRun(getContext());
+ if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
+ for (Map.Entry<String, String> entry :
Review Comment:
For `Source` plugins we are copying configuration from `InputFormatProvider`, for `Sink` plugins from `OutputFormatProvider`.
Do you suggest moving it to a separate method?
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r992030529
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
👍
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r979892010
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Yes, it would be good to clarify that.
We don't use Spark environment in CdapIO. Our goal is only to mimic all Spark classes related to Cdap Plugins in order to use the business logic.
At the moment, all the Cdap plugins we want to support use Spark 2, so we use Spark 2 here.
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990163666
##########
sdks/java/io/cdap/src/test/java/org/apache/beam/sdk/io/cdap/CdapIOTest.java:
##########
@@ -154,11 +171,45 @@ public void testReadingData() {
for (int i = 1; i < EmployeeInputFormat.NUM_OF_TEST_EMPLOYEE_RECORDS; i++) {
expected.add(KV.of(String.valueOf(i), EmployeeInputFormat.EMPLOYEE_NAME_PREFIX + i));
}
- PCollection<KV<String, String>> actual = p.apply("ReadTest", read);
+ PCollection<KV<String, String>> actual = p.apply("ReadBatchTest", read);
PAssert.that(actual).containsInAnyOrder(expected);
p.run();
}
+ @Test
+ public void testReadingDataStreaming() {
Review Comment:
Renamed
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r988739446
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Hmm, I personally don't see why it's necessary to repeat the same mistake just because it was made once?
Also IOs and runners are very different in nature. Replacing the runner is trivial, but with the Cdap IO you're better aware right from the start that it depends on Spark 2 (and Spark 3 is not (yet) supported).
In any case, I don't wanna block this PR ... just a long hanging fruit and missed opportunity in my eyes.
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r993325610
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/MappingUtils.java:
##########
@@ -33,23 +38,118 @@
import io.cdap.plugin.zendesk.source.batch.ZendeskBatchSource;
import io.cdap.plugin.zendesk.source.batch.ZendeskInputFormat;
import io.cdap.plugin.zendesk.source.batch.ZendeskInputFormatProvider;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.io.sparkreceiver.ReceiverBuilder;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.reflect.TypeToken;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.spark.streaming.receiver.Receiver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/** Util class for mapping plugins. */
public class MappingUtils {
- public static Plugin getPluginByClass(Class<?> pluginClass) {
+ private static final Logger LOG = LoggerFactory.getLogger(MappingUtils.class);
+ private static final String HUBSPOT_ID_FIELD = "vid";
+ private static final Gson GSON = new Gson();
+
+ private static final Map<
+ Class<?>, Pair<SerializableFunction<?, Long>, ReceiverBuilder<?, ? extends Receiver<?>>>>
+ REGISTERED_PLUGINS;
+
+ static {
+ REGISTERED_PLUGINS = new HashMap<>();
+ }
+
+ /** Gets a {@link Plugin} by its class. */
+ static Plugin getPluginByClass(Class<?> pluginClass) {
checkArgument(pluginClass != null, "Plugin class can not be null!");
if (pluginClass.equals(SalesforceBatchSource.class)) {
- return Plugin.create(
+ return Plugin.createBatch(
pluginClass, SalesforceInputFormat.class, SalesforceInputFormatProvider.class);
} else if (pluginClass.equals(HubspotBatchSource.class)) {
- return Plugin.create(pluginClass, HubspotInputFormat.class, HubspotInputFormatProvider.class);
+ return Plugin.createBatch(
+ pluginClass, HubspotInputFormat.class, HubspotInputFormatProvider.class);
} else if (pluginClass.equals(ZendeskBatchSource.class)) {
- return Plugin.create(pluginClass, ZendeskInputFormat.class, ZendeskInputFormatProvider.class);
+ return Plugin.createBatch(
+ pluginClass, ZendeskInputFormat.class, ZendeskInputFormatProvider.class);
} else if (pluginClass.equals(HubspotBatchSink.class)) {
- return Plugin.create(pluginClass, HubspotOutputFormat.class, SourceInputFormatProvider.class);
+ return Plugin.createBatch(
+ pluginClass, HubspotOutputFormat.class, SourceInputFormatProvider.class);
} else if (pluginClass.equals(ServiceNowSource.class)) {
- return Plugin.create(
+ return Plugin.createBatch(
pluginClass, ServiceNowInputFormat.class, SourceInputFormatProvider.class);
+ } else if (pluginClass.equals(HubspotStreamingSource.class)) {
+ return Plugin.createStreaming(pluginClass);
+ }
+ throw new UnsupportedOperationException(
Review Comment:
Yes, we already add such ability. There are two methods in `CdapIO`:
- `withCdapPluginClass()` - it will try to get `Plugin` by class from predefined list in `MappingUtils`, as we can see here.
- `withCdapPlugin()` - we can pass manually `Plugin.create(...)` with all needed parameters, it is used for example in tests. (without using `MappingUtils` class)
--
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] mosche commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1282530080
LGTM @aromanenko-dev
Though - unrelated - one more question @Lizzfox, what's the purpose / importance of Spark in the picture here? We're actually preparing the removal of the deprecated Spark2 runner (asap). The runner is badly broken and not usable anymore due to a Jackson dependency conflict (see #23568). Unfortunately this wasn't caught in local tests, but only if submitting a job to a Spark 2.4 cluster. Does this impact you?
--
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] Lizzfox commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r991577420
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Thank you all for the review and comments on the PR. Can we do this renaming you're talking about in the next PR (after this one is completed) without running into multiple conflicts?
--
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] aromanenko-dev commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r992553532
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
@Lizzfox If would be great to make these changes (I'm fine for a separate PR for this) till the next release branch cut.
--
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] Lizzfox commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1276266246
@aromanenko-dev @chamikaramj @mosche
All your comments have been addressed. Can we consider this PR ready for merge?
--
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] aromanenko-dev merged pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev merged PR #22584:
URL: https://github.com/apache/beam/pull/22584
--
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] github-actions[bot] commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1252655590
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @lukecwik for label java.
R: @Abacn for label io.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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] aromanenko-dev commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r988724793
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Unfortunately, I'm disagree with that, at least, for Spark v2 entities. We don't have this naming strategy for Spark runner (why? it's another question), so it will be confusing to have `spark2*` and `spark*` names in the same time.
Though, for Spark v3 we used to have `spark3` prefix, so we can continue that strategy for `spark4` and etc if it will be needed.
Actually, it looks as more wide question that can be applied for similar cases in other Beam components.
--
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] chamikaramj commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r989260298
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Yeah, ":sdks:java:io:sparkreceiver:2" seems to be more consistent with the current convention.
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r988715268
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
yes, modules that depends on spark should contain the spark version in their name. spark2receiver sounds good 👍
It looks like the cdap module also depends on spark, so the same applies there:
```
implementation library.java.spark_streaming
```
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990164230
##########
sdks/java/io/cdap/src/test/java/org/apache/beam/sdk/io/cdap/CdapIOTest.java:
##########
@@ -154,11 +171,45 @@ public void testReadingData() {
for (int i = 1; i < EmployeeInputFormat.NUM_OF_TEST_EMPLOYEE_RECORDS; i++) {
expected.add(KV.of(String.valueOf(i), EmployeeInputFormat.EMPLOYEE_NAME_PREFIX + i));
}
- PCollection<KV<String, String>> actual = p.apply("ReadTest", read);
+ PCollection<KV<String, String>> actual = p.apply("ReadBatchTest", read);
PAssert.that(actual).containsInAnyOrder(expected);
p.run();
}
+ @Test
+ public void testReadingDataStreaming() {
+ DirectOptions options = PipelineOptionsFactory.as(DirectOptions.class);
+ options.setBlockOnRun(false);
+ options.setRunner(DirectRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ EmployeeConfig pluginConfig =
+ new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
+ MappingUtils.registerStreamingPlugin(
+ EmployeeStreamingSource.class,
+ Long::valueOf,
+ new ReceiverBuilder<>(EmployeeReceiver.class).withConstructorArgs(pluginConfig));
+
+ CdapIO.Read<String, String> read =
+ CdapIO.<String, String>read()
+ .withCdapPlugin(Plugin.createStreaming(EmployeeStreamingSource.class))
+ .withPluginConfig(pluginConfig)
+ .withKeyClass(String.class)
+ .withValueClass(String.class);
+
+ List<String> storedRecords = EmployeeReceiver.getStoredRecords();
+ List<String> outputRecords = TestOutputDoFn.getRecords();
+ outputRecords.clear();
+
+ p.apply("ReadStreamingTest", read)
+ .setCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), StringUtf8Coder.of()))
+ .apply(Values.create())
+ .apply(ParDo.of(new TestOutputDoFn()));
+
+ p.run().waitUntilFinish(Duration.standardSeconds(15));
+ assertEquals(storedRecords, outputRecords);
Review Comment:
Changed to PAssert
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990164480
##########
sdks/java/io/cdap/src/test/java/org/apache/beam/sdk/io/cdap/TestOutputDoFn.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.sdk.transforms.DoFn;
+
+public class TestOutputDoFn extends DoFn<String, String> {
+ private static final List<String> RECORDS = new ArrayList<>();
+
+ @ProcessElement
+ public void processElement(@Element String input, DoFn.OutputReceiver<String> output) {
+ RECORDS.add(input);
Review Comment:
Removed
--
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] Amar3tto commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1271978188
Run Java_Examples_Dataflow PreCommit
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r982579185
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
@Amar3tto Unfortunately this can't be dynamically resolved. If we (eventually) want to support Spark 2 and 3, there has to be 2 maven artifacts. I was talking about a naming scheme for these.
So something like:
- org.apache.beam/beam-sdks-java-io-cdap-spark2
- org.apache.beam/beam-sdks-java-io-cdap-spark3
--
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] Amar3tto commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1271724260
Run Java_Examples_Dataflow PreCommit
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r938505868
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
Review Comment:
Done
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r978366801
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Just noticing this is based on Spark 2.4 and wondering.... considering Spark 2 is close to reaching it's end of life, wouldn't it make more sense to kick this of with Spark 3?
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r938507548
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
+ try {
+ Constructor<?> constructor =
+ getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
+ constructor.setAccessible(true);
+ cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
+ } catch (Exception e) {
+ LOG.error("Can not instantiate CDAP plugin class", e);
+ throw new IllegalStateException("Can not call prepareRun");
}
- } else {
- for (Map.Entry<String, String> entry :
- getContext().getOutputFormatProvider().getOutputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ }
+ try {
+ cdapPluginObj.prepareRun(getContext());
Review Comment:
Fixed try-catch block.
Don't think that we need a separate method for this, since our method already called `prepareRun()`.
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
Review Comment:
Done
--
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] Amar3tto commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1271720069
Run RAT PreCommit
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r988739446
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Hmm, I personally don't see why it's necessary to repeat the same mistake just because it was made once?
Also IOs and runners are very different in nature. Replacing the runner is trivial, but with the Cdap IO you're better aware right from the start that it depends on Spark 2 (and Spark 3 is not (yet) supported).
In any case, I don't wanna block this PR ... just a low hanging fruit and missed opportunity in my eyes.
--
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] Amar3tto commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1271691790
Run RAT PreCommit
--
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] aromanenko-dev commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1282483131
@chamikaramj @mosche
Do you have additional comments on this PR?
--
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] Amar3tto commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1250597220
Run Java PreCommit
--
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] mosche commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r981129510
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Should we plan ahead then and use a naming scheme that would allow to support Spark 3 later and avoids any potential confusion with regards to the supported spark version?
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r982571836
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
@mosche What do you mean by naming scheme?
Do you suggest to add something like this?
```
def shouldProvideSpark2 = true
def shouldProvideSpark3 = false
```
...
```
if (shouldProvideSpark2) {
implementation library.java.spark_streaming
implementation library.java.spark_core
}
if (shouldProvideSpark3) {
implementation library.java.spark3_streaming
implementation library.java.spark3_core
}
```
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990165758
##########
sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java:
##########
@@ -141,15 +145,42 @@ public boolean hasRecords() {
@Override
public void start(Receiver<V> sparkReceiver) {
this.sparkReceiver = sparkReceiver;
- try {
- new WrappedSupervisor(
- sparkReceiver,
- new SparkConf(),
- objects -> {
- V record = (V) objects[0];
- recordsQueue.offer(record);
+
+ final SerializableFunction<Object[], Void> storeFn =
+ (input) -> {
+ if (input == null) {
return null;
- });
+ }
+
+ if (input[0] instanceof ByteBuffer) {
+ final ByteBuffer byteBuffer = ((ByteBuffer) input[0]).asReadOnlyBuffer();
+ final byte[] bytes = new byte[byteBuffer.limit()];
+ byteBuffer.get(bytes);
+ final V record = SerializationUtils.deserialize(bytes);
+ recordsQueue.offer(record);
+ } else if (input[0] instanceof Iterator) {
+ final Iterator<V> iterator = (Iterator<V>) input[0];
+ while (iterator.hasNext()) {
+ V record = (V) iterator.next();
+ recordsQueue.offer(record);
+ }
+ } else if (input[0] instanceof ArrayBuffer) {
+ final ArrayBuffer<V> arrayBuffer = (ArrayBuffer<V>) input[0];
+ final Iterator<V> iterator = arrayBuffer.iterator();
+ while (iterator.hasNext()) {
Review Comment:
Added three new tests that cover this logic.
--
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] Lizzfox commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r992731049
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
@aromanenko-dev got it, thanks! We'll make changes as soon as this PR is merged and I'll tag you in the comments.
--
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] sorokin-andrey commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
sorokin-andrey commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r938434822
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
+ try {
+ Constructor<?> constructor =
+ getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
+ constructor.setAccessible(true);
+ cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
+ } catch (Exception e) {
+ LOG.error("Can not instantiate CDAP plugin class", e);
+ throw new IllegalStateException("Can not call prepareRun");
}
- } else {
- for (Map.Entry<String, String> entry :
- getContext().getOutputFormatProvider().getOutputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ }
+ try {
+ cdapPluginObj.prepareRun(getContext());
Review Comment:
Basically, we need this try-catch only to deal with this line of code. If you move it to a separate private method, it will increase the readability of the code dramatically.
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
Review Comment:
I would recommend inverting this conditional operator.
This will allow us to reduce the nesting of the code by 1 tab.
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
+ try {
+ Constructor<?> constructor =
+ getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
+ constructor.setAccessible(true);
+ cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
+ } catch (Exception e) {
+ LOG.error("Can not instantiate CDAP plugin class", e);
+ throw new IllegalStateException("Can not call prepareRun");
}
- } else {
- for (Map.Entry<String, String> entry :
- getContext().getOutputFormatProvider().getOutputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ }
+ try {
+ cdapPluginObj.prepareRun(getContext());
+ if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
+ for (Map.Entry<String, String> entry :
Review Comment:
Personally, I think this is unclear what you are doing here and why. Plus these for loops look almost identical, except for the collection part.
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:
##########
@@ -83,46 +86,50 @@ public Plugin withConfig(PluginConfig pluginConfig) {
* validating connection to the CDAP sink/source and performing initial tuning.
*/
public void prepareRun() {
- PluginConfig pluginConfig = getPluginConfig();
- checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
- if (cdapPluginObj == null) {
- try {
- Constructor<?> constructor =
- getPluginClass().getDeclaredConstructor(pluginConfig.getClass());
- constructor.setAccessible(true);
- cdapPluginObj = (SubmitterLifecycle) constructor.newInstance(pluginConfig);
- } catch (Exception e) {
- LOG.error("Can not instantiate CDAP plugin class", e);
- throw new IllegalStateException("Can not call prepareRun");
- }
- }
- try {
- cdapPluginObj.prepareRun(getContext());
- if (getPluginType().equals(PluginConstants.PluginType.SOURCE)) {
- for (Map.Entry<String, String> entry :
- getContext().getInputFormatProvider().getInputFormatConfiguration().entrySet()) {
- getHadoopConfiguration().set(entry.getKey(), entry.getValue());
+ if (!isUnbounded()) {
+ PluginConfig pluginConfig = getPluginConfig();
+ checkStateNotNull(pluginConfig, "PluginConfig should be not null!");
+ if (cdapPluginObj == null) {
Review Comment:
Why didn't you move this conditional block to the separate method?
--
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] Lizzfox commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1255185982
@chamikaramj @aromanenko-dev
[CdapIO] Integration CdapIO with SparkReceiverIO PR is ready for review
Thank you!
--
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] aromanenko-dev commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r978719411
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Good question, I missed that
--
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] Lizzfox commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r984790400
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
@mosche
Thank you for your comment! In the current implementation of SparkReceiverIO, we didn't plan to support Spark version 3, so at the moment we can plan a package that supports version 3 of spark streaming as you suggest.
--
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] Lizzfox commented on pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Lizzfox commented on PR #22584:
URL: https://github.com/apache/beam/pull/22584#issuecomment-1284146595
@mosche we don't use Spark runner, we only support Dataflow runner in our implementation. Thanks
--
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] aromanenko-dev commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
aromanenko-dev commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r989257387
##########
sdks/java/io/cdap/build.gradle:
##########
@@ -52,14 +52,17 @@ dependencies {
implementation library.java.cdap_plugin_zendesk
implementation library.java.commons_lang3
implementation library.java.guava
+ implementation library.java.google_code_gson
implementation library.java.hadoop_common
implementation library.java.hadoop_mapreduce_client_core
implementation library.java.jackson_core
implementation library.java.jackson_databind
implementation library.java.slf4j_api
+ implementation library.java.spark_streaming
Review Comment:
Well, then it should be `:sdks:java:io:sparkreceiver:2` (like `:runners:spark:2` and `:runners:spark:3`) but not `:sdks:java:io:spark2receiver`
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990157807
##########
sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/CdapIO.java:
##########
@@ -17,29 +17,165 @@
*/
package org.apache.beam.sdk.io.cdap;
-import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
+import static org.apache.beam.sdk.io.cdap.MappingUtils.getOffsetFnForPluginClass;
+import static org.apache.beam.sdk.io.cdap.MappingUtils.getPluginByClass;
+import static org.apache.beam.sdk.io.cdap.MappingUtils.getReceiverBuilderByPluginClass;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
import com.google.auto.value.AutoValue;
import io.cdap.cdap.api.plugin.PluginConfig;
+import java.util.Map;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.coders.CannotProvideCoderException;
+import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.io.hadoop.format.HDFSSynchronization;
import org.apache.beam.sdk.io.hadoop.format.HadoopFormatIO;
+import org.apache.beam.sdk.io.sparkreceiver.SparkReceiverIO;
+import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PDone;
+import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.OutputFormat;
import org.checkerframework.checker.nullness.qual.Nullable;
/**
- * An unbounded/bounded sources and sinks from <a
- * href="https://github.com/data-integrations">CDAP</a> plugins.
+ * A {@link CdapIO} is a Transform for reading data from source or writing data to sink of a Cdap
+ * Plugin. It uses {@link HadoopFormatIO} for Batch and SparkReceiverIO for Streaming.
+ *
+ * <h2>Read from Cdap Plugin Bounded Source</h2>
+ *
+ * <p>To configure {@link CdapIO} source, you must specify Cdap {@link Plugin}, Cdap {@link
+ * PluginConfig}, key and value classes.
+ *
+ * <p>{@link Plugin} is the Wrapper class for the Cdap Plugin. It contains main information about
+ * the Plugin. The object of the {@link Plugin} class can be created with the {@link
+ * Plugin#createBatch(Class, Class, Class)} method. Method requires the following parameters:
+ *
+ * <ul>
+ * <li>{@link io.cdap.cdap.etl.api.batch.BatchSource} class
+ * <li>{@link InputFormat} class
+ * <li>{@link io.cdap.cdap.api.data.batch.InputFormatProvider} class
+ * </ul>
+ *
+ * <p>For more information about the InputFormat and InputFormatProvider, see {@link
+ * HadoopFormatIO}.
+ *
+ * <p>Every Cdap Plugin has its {@link PluginConfig} class with necessary fields to configure the
+ * Plugin. You can set the {@link Map} of your parameters with the {@link
+ * ConfigWrapper#withParams(Map)} method where the key is the field name.
+ *
+ * <p>For example, to create a basic {@link CdapIO#read()} transform:
+ *
+ * <pre>{@code
+ * Pipeline p = ...; // Create pipeline.
+ *
+ * // Create PluginConfig for specific plugin
+ * EmployeeConfig pluginConfig =
+ * new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
+ *
+ * // Read using CDAP batch plugin
+ * p.apply("ReadBatch",
+ * CdapIO.<String, String>read()
+ * .withCdapPlugin(
+ * Plugin.createBatch(
+ * EmployeeBatchSource.class,
+ * EmployeeInputFormat.class,
+ * EmployeeInputFormatProvider.class))
+ * .withPluginConfig(pluginConfig)
+ * .withKeyClass(String.class)
+ * .withValueClass(String.class));
+ * }</pre>
+ *
+ * <h2>Write to Cdap Plugin Bounded Sink</h2>
+ *
+ * <p>To configure {@link CdapIO} sink, just as {@link CdapIO#read()} Cdap {@link Plugin}, Cdap
+ * {@link PluginConfig}, key, value classes must be specified. In addition, it's necessary to
+ * determine locks directory path {@link CdapIO.Write#withLocksDirPath(String)}. It's used for
+ * {@link HDFSSynchronization} configuration for {@link HadoopFormatIO}. More info can be found in
+ * {@link HadoopFormatIO} documentation.
+ *
+ * <p>To create the object of the {@link Plugin} class with the {@link Plugin#createBatch(Class,
+ * Class, Class)} method, need to specify the following parameters:
+ *
+ * <ul>
+ * <li>{@link io.cdap.cdap.etl.api.batch.BatchSink} class
+ * <li>{@link OutputFormat} class
+ * <li>{@link io.cdap.cdap.api.data.batch.OutputFormatProvider} class
+ * </ul>
+ *
+ * <p>For more information about the OutputFormat and OutputFormatProvider, see {@link
+ * HadoopFormatIO}.
+ *
+ * <p>Example of {@link CdapIO#write()} usage:
+ *
+ * <pre>{@code
+ * Pipeline p = ...; // Create pipeline.
+ *
+ * // Get or create data to write
+ * PCollection<KV<String, String>> input = p.apply(Create.of(data));
+ *
+ * // Create PluginConfig for specific plugin
+ * EmployeeConfig pluginConfig =
+ * new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
+ *
+ * // Write using CDAP batch plugin
+ * input.apply(
+ * "WriteBatch",
+ * CdapIO.<String, String>write()
+ * .withCdapPlugin(
+ * Plugin.createBatch(
+ * EmployeeBatchSink.class,
+ * EmployeeOutputFormat.class,
+ * EmployeeOutputFormatProvider.class))
+ * .withPluginConfig(pluginConfig)
+ * .withKeyClass(String.class)
+ * .withValueClass(String.class)
+ * .withLocksDirPath(tmpFolder.getRoot().getAbsolutePath()));
+ * p.run();
+ * }</pre>
+ *
+ * <h2>Read from Cdap Plugin Streaming Source</h2>
+ *
+ * <p>To configure {@link CdapIO} source, you must specify Cdap {@link Plugin}, Cdap {@link
+ * PluginConfig}, key and value classes.
+ *
+ * <p>{@link Plugin} is the Wrapper class for the Cdap Plugin. It contains main information about
+ * the Plugin. The object of the {@link Plugin} class can be created with the {@link
+ * Plugin#createStreaming(Class)} method. Method requires {@link
+ * io.cdap.cdap.etl.api.streaming.StreamingSource} class parameter.
+ *
+ * <p>Every Cdap Plugin has its {@link PluginConfig} class with necessary fields to configure the
+ * Plugin. You can set the {@link Map} of your parameters with the {@link
+ * ConfigWrapper#withParams(Map)} method where the key is the field name.
+ *
+ * <p>For example, to create a basic {@link CdapIO#read()} transform:
+ *
+ * <pre>{@code
+ * Pipeline p = ...; // Create pipeline.
+ *
+ * // Create PluginConfig for specific plugin
+ * EmployeeConfig pluginConfig =
+ * new ConfigWrapper<>(EmployeeConfig.class).withParams(TEST_EMPLOYEE_PARAMS_MAP).build();
+ *
+ * // Read using CDAP streaming plugin
+ * p.apply("ReadStreaming",
+ * CdapIO.<String, String>read()
+ * .withCdapPlugin(Plugin.createStreaming(EmployeeStreamingSource.class))
+ * .withPluginConfig(pluginConfig)
+ * .withKeyClass(String.class)
+ * .withValueClass(String.class));
+ * }</pre>
*/
@Experimental(Kind.SOURCE_SINK)
+@SuppressWarnings("nullness")
Review Comment:
Fixed
--
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] Amar3tto commented on a diff in pull request #22584: [CdapIO] Integration CdapIO with SparkReceiverIO
Posted by GitBox <gi...@apache.org>.
Amar3tto commented on code in PR #22584:
URL: https://github.com/apache/beam/pull/22584#discussion_r990165118
##########
sdks/java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java:
##########
@@ -141,15 +145,42 @@ public boolean hasRecords() {
@Override
public void start(Receiver<V> sparkReceiver) {
this.sparkReceiver = sparkReceiver;
- try {
- new WrappedSupervisor(
- sparkReceiver,
- new SparkConf(),
- objects -> {
- V record = (V) objects[0];
- recordsQueue.offer(record);
+
+ final SerializableFunction<Object[], Void> storeFn =
+ (input) -> {
+ if (input == null) {
return null;
- });
+ }
+
+ if (input[0] instanceof ByteBuffer) {
+ final ByteBuffer byteBuffer = ((ByteBuffer) input[0]).asReadOnlyBuffer();
Review Comment:
Added a comment:
`Use only [0] element - data.
The other elements are not needed because they are related to Spark environment options.`
--
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