Merge branch 'development'

Author: RuslanBergenov

README.md

@@ -22,7 +22,8 @@ and [Hubspot] to Google BigQuery.
         - [Partitioning background](#partitioning-background)
         - [Clustering background](#clustering-background)
         - [Setting up partitioning and clustering](#setting-up-partitioning-and-clustering)
-    - [Step 6: target-tables-config file: force data types and modes](#step-6-target-tables-config-file-force-data-types-and-modes)
+    - [Step 6 (Optional): target-tables-config file: force data types and modes](#step-6-optional-target-tables-config-file-force-data-types-and-modes)
+    - [Step 7 (Optional): target-tables-config file: rename a field](#step-7-optional-target-tables-config-file-rename-a-field)
 - [Unit tests set up](#unit-tests-set-up)
 - [Config files in this project](#config-files-in-this-project)
 
@@ -363,14 +364,11 @@ You can only set up partitioning.
 }}
 ```
 
-3. Clear you **state.json**, so it's an empty JSON `{}`, because we want to load all data again. Skip this step, if you
-   didn't previously load this data in **Step 4** above.
+3. Clear you **state.json**, so it's an empty JSON `{}`, because we want to load all data again. Skip this step, if you didn't previously load this data in **Step 4** above.
 
-4. Delete your BigQuery destination table **exchangeratesapi**, because we want to re-load it again from scratch. Skip
-   this step, if you didn't previously load this data in **Step 4** above.
+4. Delete your BigQuery destination table **exchangeratesapi**, because we want to re-load it again from scratch. Skip this step, if you didn't previously load this data in **Step 4** above.
 
-3. Load data data into BigQuery, while configuring target tables. Pass **target-tables-config.json** as a command line
-   argument.
+3. Load data data into BigQuery, while configuring target tables. Pass **target-tables-config.json** as a command line argument.
 
 ```bash
 {project_root_dir}\tap\Scripts\tap-exchangeratesapi --config sample_config/tap-config-exchange-rates-api.json | ^
@@ -379,35 +377,30 @@ You can only set up partitioning.
 ```
 
 - "^" indicates a new line in Windows Command Prompt. In Mac terminal, use "\\".
-- If you don't want to pass **target-tables-config.json** file as a CLI argument, you can
-  add ```"table_config": "target-tables-config.json"``` to your **target-config.json** file. See **Step 3: Configure**
-  above.
+- If you don't want to pass **target-tables-config.json** file as a CLI argument, you can add ```"table_config": "target-tables-config.json"``` to your **target-config.json** file. See **Step 3: Configure** above.
 
 
-6. Verify in BigQuery web UI that partitioning and clustering worked (in our example below, we only set up
-   partitioning):
+6. Verify in BigQuery web UI that partitioning and clustering worked (in our example below, we only set up partitioning):
+
 
 <img src="readme_screenshots/14_Partitioned_Table.png" width="650" alt="Download the service account credential JSON file">
 
-### Step 6: target-tables-config file: force data types and modes
+
+### Step 6 (Optional): target-tables-config file: force data types and modes
 
 #### Problem:
 
 - Normally, tap catalog file governs schema of data which will be loaded into target-bigquery.
 - However, sometimes you can get a column of an undesired data type, which is not following your tap-catalog file.
 
 #### Solution:
-
-- You can force that column to the desired data type by using `force_fields` flag inside your *
-  target-tables-config.json* file.
+- You can force that column to the desired data type by using the `force_fields` flag inside your *target-tables-config.json* file.
 
 #### Example:
-
-- We used this solution to fix `"date_start"` field from `"ads_insights_age_and_gender"` stream from tap-facebook.
-- In tap catalog file, we said we wanted this column to be a **date**.
-- However, the tap generates schema where this column is a **string**, despite our tap catalog file.
-- Therefore, we used `force_fields` flag in target-tables-config.json to override what the tap generates and force the
-  column to be a date.
+- We used this solution to fix `"date_start"` field from `"ads_insights_age_and_gender"` stream from tap-facebook. 
+- In the tap catalog file, we said we wanted this column to be a **date**. 
+- However, the tap generates schema where this column is a **string**, despite our tap catalog file. 
+- Therefore, we used `force_fields` flag in target-tables-config.json to override what the tap generates and force the column to be a date.
 - Example of *target-tables-config.json* file:
 
 ```
@@ -425,6 +418,23 @@ You can only set up partitioning.
 }
 ```
 
+### Step 7 (Optional): target-tables-config file: rename a field
+
+#### Problem and solution:
+- You can rename a field, using the `force_fields` flag inside your *target-tables-config.json* file.
+  
+#### Example:
+- Example of *target-tables-config.json* file where we renamed a field from `old_name` to `new_name`:
+```
+      "ads_insights_age_and_gender": {
+        "partition_field": "date_start",
+        "cluster_fields": ["age", "gender","account_id", "campaign_id"],
+        "force_fields": {
+          "old_name":{"bq_field_name": "new_name"}
+        }
+      }
+```
+
 ## Unit tests set up
 
 Add the following files to *sandbox* directory under project root directory:

target_bigquery/processhandler.py

@@ -179,7 +179,7 @@ def handle_record_message(self, msg):
 
         schema = self.schemas[stream]
         bq_schema = self.bq_schema_dicts[stream]
-        nr = cleanup_record(schema, msg.record)
+        nr = cleanup_record(schema, msg.record, force_fields=self.table_configs.get(msg.stream, {}).get("force_fields", {}))
 
         try:
             nr = format_record_to_schema(nr, self.bq_schema_dicts[stream])
@@ -188,6 +188,8 @@ def handle_record_message(self, msg):
             self.logger.critical(f"Cannot format a record for stream {msg.stream} to its corresponding BigQuery schema. Details: {extra}")
             raise e
 
+
+
         # schema validation may fail if data doesn't match schema in terms of data types
         # in this case, we validate schema again on data which has been forced to match schema
         # nr is based on msg.record, but all data from msg.record has been forced to match schema

target_bigquery/schema.py

@@ -14,12 +14,18 @@
 }
 
 
-def cleanup_record(schema, record):
+def cleanup_record(schema, record, force_fields={}):
     """
     Clean up / prettify field names, make sure they match BigQuery naming conventions.
 
-    :param JSON record generated by the tap and piped into target-bigquery
-    :param bq_schema: JSON schema generated by the tap and piped into target-bigquery
+    :param schema: JSON schema generated by the tap and piped into target-bigquery
+    :param record: JSON record generated by the tap and piped into target-bigquery
+    :param force_fields: You can force a field to a desired data type via force_fields flag.
+        Use case example:
+            tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+            You can also rename a field using the force_fields parameter.
+        Please see README for more information and examples.
     :return: JSON record/data, where field names are cleaned up / prettified.
     """
     if not isinstance(record, dict) and not isinstance(record, list):
@@ -28,21 +34,21 @@ def cleanup_record(schema, record):
     elif isinstance(record, list):
         nr = []
         for item in record:
-            nr.append(cleanup_record(schema, item))
+            nr.append(cleanup_record(schema, item, force_fields))
         return nr
 
     elif isinstance(record, dict):
         nr = {}
         for key, value in record.items():
-            nkey = create_valid_bigquery_field_name(key)
-            nr[nkey] = cleanup_record(schema, value)
+            nkey = create_valid_bigquery_field_name(key, force_fields)
+            nr[nkey] = cleanup_record(schema, value, force_fields)
         return nr
 
     else:
         raise Exception(f"unhandled instance of record: {record}")
 
 
-def create_valid_bigquery_field_name(field_name):
+def create_valid_bigquery_field_name(field_name, force_fields={}):
     """
     Clean up / prettify field names, make sure they match BigQuery naming conventions.
 
@@ -56,9 +62,17 @@ def create_valid_bigquery_field_name(field_name):
             -underscore, and
         • be at most 300 characters long
 
-    :param key: JSON field name
+    :param field_name: JSON field name
+    :param force_fields: You can force a field to a desired data type via force_fields flag.
+        Use case example:
+            tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+            You can also rename a field using the force_fields parameter.
+        Please see README for more information and examples.
     :return: cleaned up JSON field name
     """
+    if field_name in force_fields and force_fields[field_name].get("bq_field_name"):
+        return force_fields[field_name]["bq_field_name"]
 
     cleaned_up_field_name = ""
 
@@ -150,12 +164,18 @@ def prioritize_one_data_type_from_multiple_ones_in_any_of(field_property):
     return min(any_of_data_types, key=any_of_data_types.get)
 
 
-def convert_field_type(field_property):
+def convert_field_type(field_name, field_property, force_fields={}):
     """
+    :param field_name: field/column name
     :param field_property: JSON field property
+    :param force_fields: You can force a field to a desired data type via force_fields flag.
+        Use case example:
+            tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+            You can also rename a field using the force_fields parameter.
+        Please see README for more information and examples.
     :return: BigQuery SchemaField field_type
     """
-
     conversion_dict = {"string": "STRING",
                        "number": "FLOAT",
                        "integer": "INTEGER",
@@ -170,7 +190,10 @@ def convert_field_type(field_property):
                        "bq-bigdecimal": "BIGDECIMAL"
                        }
 
-    if "anyOf" in field_property:
+    if field_name in force_fields and force_fields[field_name].get("type"):
+        return force_fields[field_name]["type"]
+
+    elif "anyOf" in field_property:
 
         prioritized_data_type = prioritize_one_data_type_from_multiple_ones_in_any_of(field_property)
 
@@ -208,13 +231,22 @@ def convert_field_type(field_property):
     return field_type_bigquery
 
 
-def determine_field_mode(field_property):
+def determine_field_mode(field_name, field_property, force_fields={}):
     """
     :param field_name: one nested JSON field name
     :param field_property: one nested JSON field property
+    :param force_fields: You can force a field to a desired data type via force_fields flag.
+            Use case example:
+                tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                    which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+                You can also rename a field using the force_fields parameter.
+            Please see README for more information and examples.
     :return: BigQuery SchemaField mode
     """
-    if "items" in field_property:
+    if field_name in force_fields and force_fields[field_name].get("mode"):
+        return force_fields[field_name]["mode"]
+
+    elif "items" in field_property:
 
         field_mode = 'REPEATED'
 
@@ -226,6 +258,7 @@ def determine_field_mode(field_property):
 
 
 def replace_nullable_mode_with_required(schema_field_input):
+
     schema_field_updated = SchemaField(name=schema_field_input.name,
                                        field_type=schema_field_input.field_type,
                                        mode='REQUIRED',
@@ -283,24 +316,30 @@ def determine_precision_and_scale_for_decimal_or_bigdecimal(field_property):
     return precision, scale
 
 
-def build_field(field_name, field_property):
+def build_field(field_name, field_property, force_fields):
     """
     :param field_name: one nested JSON field name
     :param field_property: one nested JSON field property
+    :param force_fields: You can force a field to a desired data type via force_fields flag.
+            Use case example:
+                tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                    which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+                You can also rename a field using the force_fields parameter.
+            Please see README for more information and examples.
     :return: one BigQuery nested SchemaField
     """
 
     if not ("items" in field_property and "properties" in field_property["items"]) and not (
             "properties" in field_property):
 
-        field_type = convert_field_type(field_property)
+        field_type = convert_field_type(field_name, field_property, force_fields)
 
         precision, scale = determine_precision_and_scale_for_decimal_or_bigdecimal(field_property) if field_type in [
             "DECIMAL", "BIGDECIMAL"] else (None, None)
 
-        return (SchemaField(name=create_valid_bigquery_field_name(field_name),
+        return (SchemaField(name=create_valid_bigquery_field_name(field_name,force_fields) ,
                             field_type=field_type,
-                            mode=determine_field_mode(field_property),
+                            mode=determine_field_mode(field_name, field_property, force_fields),
                             description=None,
                             fields=(),
                             policy_tags=None,
@@ -313,7 +352,7 @@ def build_field(field_name, field_property):
 
         processed_subfields = []
 
-        field_type = convert_field_type(field_property)
+        field_type = convert_field_type(field_name, field_property, force_fields)
 
         precision, scale = determine_precision_and_scale_for_decimal_or_bigdecimal(field_property) if field_type in [
             "DECIMAL", "BIGDECIMAL"] else (None, None)
@@ -322,11 +361,11 @@ def build_field(field_name, field_property):
         for subfield_name, subfield_property in field_property.get("properties",
                                                                    field_property.get("items", {}).get("properties")
                                                                    ).items():
-            processed_subfields.append(build_field(subfield_name, subfield_property))
+            processed_subfields.append(build_field(subfield_name, subfield_property, force_fields))
 
-        return (SchemaField(name=create_valid_bigquery_field_name(field_name),
+        return (SchemaField(name=create_valid_bigquery_field_name(field_name, force_fields),
                             field_type=field_type,
-                            mode=determine_field_mode(field_property),
+                            mode=determine_field_mode(field_name, field_property, force_fields),
                             description=None,
                             fields=processed_subfields,
                             policy_tags=None,
@@ -343,8 +382,10 @@ def build_schema(schema, key_properties=None, add_metadata=True, force_fields={}
     :param add_metadata: do we want BigQuery metadata columns (e.g., when data was uploaded?)
     :param force_fields: You can force a field to a desired data type via force_fields flag.
             Use case example:
-            tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
-            which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+                tap facebook field "date_start" from stream ads_insights_age_and_gender is being passed as string to BQ,
+                    which contradicts tap catalog file, where we said it's a date. force_fields fixes this issue.
+                You can also rename a field using the force_fields parameter.
+            Please see README for more information and examples.
     :return: a list of BigQuery SchemaFields, which represents one BigQuery table
     """
 
@@ -356,20 +397,10 @@ def build_schema(schema, key_properties=None, add_metadata=True, force_fields={}
 
     for field_name, field_property in schema.get("properties", schema.get("items", {}).get("properties", {})).items():
 
-        if field_name in force_fields:
-
-            next_field = (
-                SchemaField(field_name, force_fields[field_name]["type"],
-                            force_fields[field_name].get("mode", "nullable"),
-                            force_fields[field_name].get("description", None), ())
-            )
-
-        else:
-
-            next_field = build_field(field_name, field_property)
+        next_field = build_field(field_name, field_property, force_fields)
 
-            if field_name in required_fields:
-                next_field = replace_nullable_mode_with_required(next_field)
+        if field_name in required_fields and field_name not in force_fields:
+            next_field = replace_nullable_mode_with_required(next_field)
 
         schema_bigquery.append(next_field)
 
@@ -396,7 +427,7 @@ def format_record_to_schema(record, bq_schema):
 
     RECORD is not included into conversion_dict - it is done on purpose. RECORD is handled recursively.
 
-    :param JSON record generated by the tap and piped into target-bigquery
+    :param record: JSON record generated by the tap and piped into target-bigquery
     :param bq_schema: JSON schema generated by the tap and piped into target-bigquery
     :return: JSON record/data, where the data types match JSON schema
     """

tests/rsc/config/facebook_stream_tables_config.json

@@ -33,7 +33,9 @@
         "cluster_fields": ["age", "gender","account_id", "campaign_id"],
         "force_fields": {
           "date_start": {"type": "DATE", "mode":  "NULLABLE"},
-          "date_stop": {"type": "DATE", "mode":  "NULLABLE"}
+          "date_stop": {"type": "DATE", "mode":  "NULLABLE"},
+          "1d_click":{"bq_field_name": "lower_level_field_renamed"},
+          "unique_actions":{"bq_field_name": "parent_field_renamed"}
         }
       },
       "ads_insights_platform_and_device": {