Add SendGrid Sink Plugin

Author: dgrinenko

checkstyle.xml

@@ -13,8 +13,8 @@
   -->
 
 <!DOCTYPE module PUBLIC
-        "-//Puppy Crawl//DTD Check Configuration 1.3//EN"
-        "http://www.puppycrawl.com/dtds/configuration_1_3.dtd">
+        "-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
+        "https://checkstyle.org/dtds/configuration_1_3.dtd">
 
 <!-- This is a checkstyle configuration file. For descriptions of
 what the following rules do, please see the checkstyle configuration

docs/SendGrid-batchsink.md

@@ -34,6 +34,17 @@
         <common.logging.version>1.2</common.logging.version>
     </properties>
 
+    <scm>
+        <connection>scm:git:https://github.com/data-integrations/sendgrid.git</connection>
+        <developerConnection>scm:git@github.com:data-integrations/sendgrid.git</developerConnection>
+        <url>https://github.com/data-integrations/sendgrid</url>
+        <tag>HEAD</tag>
+    </scm>
+
+    <issueManagement>
+        <url>https://issues.cask.co/browse/CDAP</url>
+    </issueManagement>
+
     <repositories>
         <repository>
             <id>sonatype</id>

icons/SendGrid-batchsink.png

@@ -0,0 +1,72 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.sendgrid.batch.sink;
+
+import io.cdap.plugin.sendgrid.common.objects.mail.SendGridMail;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+/**
+ * An OutputFormat that sends the output of a Hadoop job to the SendGrid record writer, also
+ * it defines the output committer.
+ */
+public class SendGridOutputFormat extends OutputFormat<NullWritable, SendGridMail> {
+  @Override
+  public RecordWriter<NullWritable, SendGridMail> getRecordWriter(TaskAttemptContext taskAttemptContext) {
+    return new SendGridRecordWriter(taskAttemptContext);
+  }
+
+  @Override
+  public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {
+    // no-op
+  }
+
+  @Override
+  public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) {
+    return new OutputCommitter() {
+      @Override
+      public void setupJob(JobContext jobContext) throws IOException {
+
+      }
+
+      @Override
+      public void setupTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+      }
+
+      @Override
+      public boolean needsTaskCommit(TaskAttemptContext taskAttemptContext) throws IOException {
+        return false;
+      }
+
+      @Override
+      public void commitTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+      }
+
+      @Override
+      public void abortTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+      }
+    };
+  }
+}

pom.xml

@@ -0,0 +1,48 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.sendgrid.batch.sink;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import io.cdap.cdap.api.data.batch.OutputFormatProvider;
+
+import java.util.Map;
+
+/**
+ * Provides {@link SendGridOutputFormat} class name and configuration
+ */
+public class SendGridOutputFormatProvider implements OutputFormatProvider {
+  public static final String PROPERTY_CONFIG_JSON = "cdap.sendgrid.config";
+  private static final Gson gson = new GsonBuilder().create();
+  private final Map<String, String> conf;
+
+  SendGridOutputFormatProvider(SendGridSinkConfig config) {
+    this.conf = new ImmutableMap.Builder<String, String>()
+        .put(PROPERTY_CONFIG_JSON, gson.toJson(config))
+        .build();
+  }
+
+  @Override
+  public String getOutputFormatClassName() {
+    return SendGridOutputFormat.class.getName();
+  }
+
+  @Override
+  public Map<String, String> getOutputFormatConfiguration() {
+    return conf;
+  }
+}

src/main/java/io/cdap/plugin/sendgrid/batch/sink/SendGridOutputFormat.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.sendgrid.batch.sink;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import io.cdap.plugin.sendgrid.common.SendGridClient;
+import io.cdap.plugin.sendgrid.common.objects.mail.SendGridMail;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+
+import java.io.IOException;
+
+/**
+ * Writes {@link SendGridMail} into batches and submit them to SendGrid send API
+ */
+public class SendGridRecordWriter extends RecordWriter<NullWritable, SendGridMail> {
+  private static final Gson gson = new GsonBuilder().create();
+  private SendGridClient client;
+
+  public SendGridRecordWriter(TaskAttemptContext taskAttemptContext) {
+    Configuration conf = taskAttemptContext.getConfiguration();
+    String serializedConfig = conf.get(SendGridOutputFormatProvider.PROPERTY_CONFIG_JSON);
+    SendGridSinkConfig sgConfig  = gson.fromJson(serializedConfig, SendGridSinkConfig.class);
+
+    client = new SendGridClient(sgConfig);
+  }
+
+  @Override
+  public void write(NullWritable nullWritable, SendGridMail sendGridMail) throws IOException {
+    client.sendMail(sendGridMail);
+  }
+
+  @Override
+  public void close(TaskAttemptContext taskAttemptContext) {
+    // no-op
+  }
+}

src/main/java/io/cdap/plugin/sendgrid/batch/sink/SendGridOutputFormatProvider.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.sendgrid.batch.sink;
+
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.api.data.batch.Output;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.api.dataset.lib.KeyValue;
+import io.cdap.cdap.etl.api.Emitter;
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.cdap.etl.api.PipelineConfigurer;
+import io.cdap.cdap.etl.api.batch.BatchSink;
+import io.cdap.cdap.etl.api.batch.BatchSinkContext;
+import io.cdap.plugin.common.IdUtils;
+import io.cdap.plugin.common.LineageRecorder;
+import io.cdap.plugin.sendgrid.common.config.BaseConfig;
+import io.cdap.plugin.sendgrid.common.objects.mail.SendGridMail;
+import org.apache.hadoop.io.NullWritable;
+
+import java.util.stream.Collectors;
+
+/**
+ * Batch Sink Plugin
+ */
+@Plugin(type = BatchSink.PLUGIN_TYPE)
+@Name(BaseConfig.PLUGIN_NAME)
+@Description("Sends mails via SendGrid")
+public class SendGridSink extends BatchSink<StructuredRecord, NullWritable, SendGridMail> {
+
+  private final SendGridSinkConfig config;
+
+  public SendGridSink(SendGridSinkConfig config) {
+    this.config = config;
+  }
+
+  @Override
+  @SuppressWarnings("ThrowableNotThrown")
+  public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
+    FailureCollector failureCollector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
+
+    IdUtils.validateReferenceName(config.referenceName, failureCollector);
+
+    config.validate(failureCollector);
+    config.validate(pipelineConfigurer.getStageConfigurer().getInputSchema());
+
+    failureCollector.getOrThrowException();
+  }
+
+  @Override
+  public void prepareRun(BatchSinkContext batchSinkContext) {
+    Schema inputSchema = batchSinkContext.getInputSchema();
+    config.validate(inputSchema);
+
+    batchSinkContext.addOutput(Output.of(config.referenceName, new SendGridOutputFormatProvider(config)));
+
+    LineageRecorder lineageRecorder = new LineageRecorder(batchSinkContext, config.referenceName);
+    lineageRecorder.createExternalDataset(inputSchema);
+
+    if (inputSchema.getFields() != null && !inputSchema.getFields().isEmpty()) {
+      String operationDescription = String.format("Wrote to SendGrid %s", config.getFrom());
+      lineageRecorder.recordWrite("Write", operationDescription,
+          inputSchema.getFields().stream()
+              .map(Schema.Field::getName)
+              .collect(Collectors.toList()));
+    }
+  }
+
+  @Override
+  public void transform(StructuredRecord record, Emitter<KeyValue<NullWritable, SendGridMail>> emitter) {
+    SendGridMail sendGridMail = SendGridSinkTransformer.transform(config, record);
+    emitter.emit(new KeyValue<>(null, sendGridMail));
+  }
+}