Merge branch 'master' into create_delete_table

Author: DamiaoCastro

AzureFunctions.Extensions.GoogleBigQuery/AzureFunctions.Extensions.GoogleBigQuery.csproj

@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <Version>0.0.9.6</Version>
+    <Version>0.0.9.7</Version>
     <Authors>Damiao Castro</Authors>
     <Company>fowet.com</Company>
     <Description>Extension for Google BigQuery to work with Azure Functions and Azure Webjobs.

AzureFunctions.Extensions.GoogleBigQuery/BigQueryInsertRowService.cs

@@ -1,16 +1,17 @@
 using Google.Cloud.BigQuery.V2;
+using Newtonsoft.Json;
 using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
 
 namespace AzureFunctions.Extensions.GoogleBigQuery {
-    public class BigQueryInsertRowService {
+    internal class BigQueryInsertRowService {
 
         private const string BigQueryDateTimeFormat = "yyyy-MM-dd HH:mm:ss";
         private static System.Globalization.CultureInfo cultureUS = new System.Globalization.CultureInfo("en-US");
 
-        public static BigQueryInsertRow GetBigQueryInsertRow(GoogleBigQueryRow row, IDictionary<string, IEnumerable<PropertyInfo>> dictionaryOfProperties) {
+        internal static BigQueryInsertRow GetBigQueryInsertRow(GoogleBigQueryRow row, IDictionary<string, IEnumerable<PropertyInfo>> dictionaryOfProperties) {
             if (row == null) { throw new System.ArgumentNullException(nameof(row)); }
             if (dictionaryOfProperties == null) { throw new System.ArgumentNullException(nameof(dictionaryOfProperties)); }
 
@@ -135,6 +136,18 @@ private static object GetArrayFromEnumreable(IDictionary<string, IEnumerable<Pro
             }
         }
 
+        internal static IEnumerable<string> GetBigQueryJobLines(IEnumerable<GoogleBigQueryRow> rows)
+        {
+
+            var jsonSerializerSettings = new JsonSerializerSettings()
+            {
+                DateFormatString = BigQueryDateTimeFormat,
+                Culture = cultureUS
+            };
+
+            return rows.Select(r => JsonConvert.SerializeObject(r, jsonSerializerSettings));
+        }
+
         private static BigQueryInsertRow[] GetSubEntitiesBigQueryInsertRows(IDictionary<string, IEnumerable<PropertyInfo>> dictionaryOfProperties, IEnumerable<object> objs) {
 
             if (objs.Count() > 0) {

AzureFunctions.Extensions.GoogleBigQuery/BigQueryService.cs

@@ -20,8 +20,39 @@ public BigQueryService(GoogleBigQueryAttribute googleBigQueryAttribute, Type ite
             (this.tableSchema, this.dictionaryOfProperties) = TableSchemaBuilderService.GetTableSchema(itemType);
         }
 
+        public Task CreateTable(bool timePartitioning, CancellationToken cancellationToken) {
+
+            BigQueryClient client = GetBiqQueryClient();
+
+            return client.GetOrCreateTableAsync(
+                    googleBigQueryAttribute.DatasetId,
+                    googleBigQueryAttribute.TableId,
+                    tableSchema,
+                    null,
+                    timePartitioning ? new CreateTableOptions() { TimePartitioning = new TimePartitioning() { Type = "DAY" } } : null,
+                    cancellationToken);
+
+        }
+
+        public Task DeleteTable(CancellationToken cancellationToken) {
+
+            BigQueryClient client = GetBiqQueryClient();
+
+            return client.DeleteTableAsync(
+                    googleBigQueryAttribute.DatasetId,
+                    googleBigQueryAttribute.TableId,
+                    null,
+                    cancellationToken);
+
+        }
+
         private Task<BigQueryTable> GetTable(DateTime date, CancellationToken cancellationToken) {
+            BigQueryClient client = GetBiqQueryClient();
 
+            return client.GetTableAsync(googleBigQueryAttribute.DatasetId, $"{googleBigQueryAttribute.TableId}${date:yyyyMMdd}", null, cancellationToken);
+        }
+
+        private BigQueryClient GetBiqQueryClient() {
             GoogleCredential googleCredential = null;
             if (googleBigQueryAttribute.Credentials != null) {
                 googleCredential = GoogleCredential.FromStream(new System.IO.MemoryStream(googleBigQueryAttribute.Credentials));
@@ -34,36 +65,49 @@ private Task<BigQueryTable> GetTable(DateTime date, CancellationToken cancellati
                 }
             }
             var client = BigQueryClient.Create(googleBigQueryAttribute.ProjectId, googleCredential);
-
-            //return client.GetOrCreateTableAsync(datasetId, tableId, tableSchema, null, new CreateTableOptions() { TimePartitioning = new TimePartitioning() { Type = "DAY" } }, cancellationToken)
-            //            .ContinueWith((createTableTask) => {
-            //                return client.GetTableAsync(datasetId, $"{tableId}${date:yyyyMMdd}", null, cancellationToken);
-            //            }, cancellationToken).Unwrap();
-            return client.GetTableAsync(googleBigQueryAttribute.DatasetId, $"{googleBigQueryAttribute.TableId}${date:yyyyMMdd}", null, cancellationToken);
+            return client;
         }
 
         public Task InsertRowsAsync(DateTime date, IEnumerable<GoogleBigQueryRow> rows, CancellationToken cancellationToken) {
 
             if (rows != null && rows.Count() > 0) {
                 int dateDiff = (date - DateTime.UtcNow.Date).Days;
 
-                if (dateDiff >= -31 && dateDiff <= 16) {
+                if (dateDiff >= -31 && dateDiff <= 16)
+                {
 
                     var bigQueryRows = rows.Select(c => BigQueryInsertRowService.GetBigQueryInsertRow(c, dictionaryOfProperties));
 
                     return GetTable(date, cancellationToken)
-                        .ContinueWith((tableTask) => {
+                        .ContinueWith((tableTask) =>
+                        {
                             BigQueryTable table = tableTask.Result;
 
                             return table.InsertRowsAsync(bigQueryRows, new InsertOptions() { AllowUnknownFields = true }, cancellationToken)
-                                        .ContinueWith((insertRowsTask) => {
-                                            if (insertRowsTask.IsFaulted) {
+                                        .ContinueWith((insertRowsTask) =>
+                                        {
+                                            if (insertRowsTask.IsFaulted)
+                                            {
                                                 throw insertRowsTask.Exception.InnerExceptions.First();
                                             }
                                         });
                         }, cancellationToken).Unwrap();
 
                 }
+                else {
+
+                    BigQueryClient client = GetBiqQueryClient();
+
+                    IEnumerable<string> lines = BigQueryInsertRowService.GetBigQueryJobLines(rows);
+
+                    return client.UploadJsonAsync(
+                            googleBigQueryAttribute.DatasetId,
+                            $"{googleBigQueryAttribute.TableId}${date:yyyyMMdd}",
+                            tableSchema,
+                            lines,
+                            new UploadJsonOptions() { AllowUnknownFields = true },
+                            cancellationToken);
+                }
             }
 
             return Task.WhenAll();