Merge pull request #43 from data-integrations/bugfix_release/fix-gcs-sink

Fix path macro in gcs sink

Author: CuriousVini

src/main/java/co/cask/gcp/bigquery/BigQuerySink.java

@@ -30,6 +30,9 @@
 import co.cask.cdap.etl.api.batch.BatchSink;
 import co.cask.cdap.etl.api.batch.BatchSinkContext;
 import co.cask.hydrator.common.LineageRecorder;
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQueryException;
+import com.google.cloud.bigquery.DatasetInfo;
 import com.google.cloud.bigquery.Field;
 import com.google.cloud.bigquery.FieldList;
 import com.google.cloud.bigquery.LegacySQLTypeName;
@@ -92,6 +95,16 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
 
   @Override
   public void prepareRun(BatchSinkContext context) throws Exception {
+    BigQuery bigquery = BigQueryUtils.getBigQuery(config.getServiceAccountFilePath(), config.getProject());
+    // create dataset if dataset does not exist
+    if (bigquery.getDataset(config.dataset) == null) {
+      try {
+        bigquery.create(DatasetInfo.newBuilder(config.dataset).build());
+      } catch (BigQueryException e) {
+        throw new RuntimeException("Exception occured while creating dataset " + config.dataset + ".", e);
+      }
+    }
+
     validateSchema();
 
     uuid = UUID.randomUUID();

src/main/java/co/cask/gcp/bigquery/BigQueryUtils.java

@@ -116,17 +116,25 @@ static Configuration getBigQueryConfig(@Nullable String serviceAccountFilePath,
   @Nullable
   static Table getBigQueryTable(@Nullable String serviceAccountFilePath, String project,
                                 String dataset, String table) throws IOException {
-    BigQuery bigquery;
+    BigQuery bigquery = getBigQuery(serviceAccountFilePath, project);
+
+    TableId id = TableId.of(project, dataset, table);
+    return bigquery.getTable(id);
+  }
+
+  /**
+   * Get BigQuery service
+   * @param serviceAccountFilePath service account file path
+   * @param project BigQuery project ID
+   */
+  static BigQuery getBigQuery(@Nullable String serviceAccountFilePath, String project) throws IOException {
     BigQueryOptions.Builder bigqueryBuilder = BigQueryOptions.newBuilder();
     if (serviceAccountFilePath != null) {
       bigqueryBuilder.setCredentials(loadServiceAccountCredentials(serviceAccountFilePath));
     }
 
     bigqueryBuilder.setProjectId(project);
-    bigquery = bigqueryBuilder.build().getService();
-
-    TableId id = TableId.of(project, dataset, table);
-    return bigquery.getTable(id);
+    return bigqueryBuilder.build().getService();
   }
 
   /**

src/main/java/co/cask/gcp/gcs/sink/GCSBatchSink.java

@@ -116,7 +116,10 @@ public static class GCSBatchSinkConfig extends GCPReferenceSinkConfig implements
     @Override
     public void validate() {
       super.validate();
-      GCSConfigHelper.getPath(path);
+      // validate that path is valid
+      if (!containsMacro("path")) {
+        GCSConfigHelper.getPath(path);
+      }
       if (suffix != null && !containsMacro("suffix")) {
         new SimpleDateFormat(suffix);
       }