Map JSON schema to BigQuery schema

This makes some assumptions about the content of repeated fields, but as
we control the event schema we think it is unlikely to break
unexpectedly

lib/dfe/analytics/event_schema.rb

@@ -5,6 +5,69 @@ def self.as_json
         path = "#{Gem.loaded_specs['dfe-analytics'].gem_dir}/config/event-schema.json"
         File.read(path)
       end
+
+      def self.as_bigquery_schema
+        schema = JSON.parse(as_json)
+        required_fields = schema['required']
+
+        properties = schema['properties']
+
+        schema = properties.keys.reduce([]) do |bq_schema, json_schema_entry_name|
+          json_schema_entry = properties[json_schema_entry_name]
+          bigquery_field_type = resolve_bigquery_type(json_schema_entry)
+
+          bigquery_schema_entry = {
+            'mode' => resolve_bigquery_mode(json_schema_entry_name, json_schema_entry, required_fields),
+            'name' => json_schema_entry_name,
+            'type' => bigquery_field_type
+          }
+
+          if bigquery_field_type == 'RECORD'
+            bigquery_schema_entry['fields'] = [
+              {
+                'mode' => 'REQUIRED',
+                'name' => 'key',
+                'type' => 'STRING'
+              },
+              {
+                'mode' => 'REPEATED',
+                'name' => 'value',
+                'type' => 'STRING'
+              }
+            ]
+          end
+
+          bq_schema << bigquery_schema_entry
+          bq_schema
+        end
+
+        schema.to_json
+      end
+
+      def self.resolve_bigquery_mode(json_schema_entry_name, json_schema_entry, required_fields)
+        if required_fields.include?(json_schema_entry_name)
+          'REQUIRED'
+        elsif json_schema_entry['type'] == 'array'
+          'REPEATED'
+        else
+          'NULLABLE'
+        end
+      end
+
+      def self.resolve_bigquery_type(json_schema_entry)
+        json_type = json_schema_entry['type']
+        json_format = json_schema_entry['format']
+
+        if json_type == 'array'
+          'RECORD'
+        elsif json_type == 'string' && json_format == 'date-time'
+          'TIMESTAMP'
+        elsif json_type == 'string'
+          'STRING'
+        elsif json_type == 'integer'
+          'INTEGER'
+        end
+      end
     end
   end
 end

lib/dfe/analytics/tasks/schema.rake

@@ -2,8 +2,12 @@ namespace :dfe do
   namespace :analytics do
     desc 'Print out the dfe-analytics JSON schema'
     task :schema do
-      path = "#{Gem.loaded_specs['dfe-analytics'].gem_dir}/config/event-schema.json"
-      puts File.read(path)
+      puts DfE::Analytics::EventSchema.as_json
+    end
+
+    desc 'Print out the dfe-analytics BigQuery schema'
+    task :big_query_schema do
+      puts DfE::Analytics::EventSchema.as_bigquery_schema
     end
   end
 end

spec/dfe/analytics/event_schema_spec.rb

@@ -9,4 +9,15 @@
       expect(output).to eq schema_on_disk
     end
   end
+
+  describe '.as_bigquery_schema' do
+    it 'transforms the JSON schema into a BQ schema' do
+      bq_schema_on_disk = File.read('spec/examples/bigquery_schema.json')
+
+      output = JSON.parse(described_class.as_bigquery_schema)
+
+      expect(output).to be_present
+      expect(output).to match_array JSON.parse(bq_schema_on_disk)
+    end
+  end
 end

spec/examples/bigquery_schema.json

@@ -0,0 +1,106 @@
+[
+  {
+    "mode": "REQUIRED",
+    "name": "occurred_at",
+    "type": "TIMESTAMP"
+  },
+  {
+    "mode": "REQUIRED",
+    "name": "event_type",
+    "type": "STRING"
+  },
+  {
+    "mode": "REQUIRED",
+    "name": "environment",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "namespace",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "user_id",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "request_uuid",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "request_method",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "request_path",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "request_user_agent",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "request_referer",
+    "type": "STRING"
+  },
+  {
+    "fields": [
+      {
+        "mode": "REQUIRED",
+        "name": "key",
+        "type": "STRING"
+      },
+      {
+        "mode": "REPEATED",
+        "name": "value",
+        "type": "STRING"
+      }
+    ],
+    "mode": "REPEATED",
+    "name": "request_query",
+    "type": "RECORD"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "response_content_type",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "response_status",
+    "type": "STRING"
+  },
+  {
+    "fields": [
+      {
+        "mode": "REQUIRED",
+        "name": "key",
+        "type": "STRING"
+      },
+      {
+        "mode": "REPEATED",
+        "name": "value",
+        "type": "STRING"
+      }
+    ],
+    "mode": "REPEATED",
+    "name": "data",
+    "type": "RECORD"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "entity_table_name",
+    "type": "STRING"
+  },
+  {
+    "mode": "NULLABLE",
+    "name": "anonymised_user_agent_and_ip",
+    "type": "STRING"
+  }
+]