Merge pull request #60 from google/push_ads_api_to_v10

Push ads api to v10

Author: antoniolmm

.github/workflows/python-app.yml

@@ -24,6 +24,7 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install -r megalista_dataflow/requirements.txt
+        pip install --no-deps -r megalista_dataflow/requirements-no-deps.txt
     - name: Run tests
       run: |
         ./run_tests.sh

megalista_dataflow/main.py

@@ -25,7 +25,7 @@
 from models.oauth_credentials import OAuthCredentials
 from models.options import DataflowOptions
 from models.sheets_config import SheetsConfig
-from sources.batches_from_executions import BatchesFromExecutions, ExecutionCoder
+from sources.batches_from_executions import BatchesFromExecutions, ExecutionCoder, TransactionalType
 from sources.primary_execution_source import PrimaryExecutionSource
 from uploaders.big_query.transactional_events_results_writer import TransactionalEventsResultsWriter
 from uploaders.campaign_manager.campaign_manager_conversion_uploader import CampaignManagerConversionUploaderDoFn
@@ -155,14 +155,24 @@ def expand(self, executions):
         return (
             executions
             | "Load Data - GoogleAdsOfflineConversions"
-            >> BatchesFromExecutions(DestinationType.ADS_OFFLINE_CONVERSION, 2000)
+            >> BatchesFromExecutions(
+                DestinationType.ADS_OFFLINE_CONVERSION,
+                2000,
+                TransactionalType.GCLID_TIME,
+                self.params.dataflow_options.bq_ops_dataset)
             | "Upload - GoogleAdsOfflineConversions"
             >> beam.ParDo(
                 GoogleAdsOfflineUploaderDoFn(
                     self.params._oauth_credentials,
                     self.params._dataflow_options.developer_token
                 )
             )
+            | "Persist results - GoogleAdsOfflineConversions"
+            >> beam.ParDo(
+              TransactionalEventsResultsWriter(
+                self.params._dataflow_options.bq_ops_dataset,
+                TransactionalType.GCLID_TIME)
+            )
         )
 
 
@@ -200,13 +210,15 @@ def expand(self, executions):
             >> BatchesFromExecutions(
                 DestinationType.GA_MEASUREMENT_PROTOCOL,
                 20,
-                True,
+                TransactionalType.UUID,
                 self.params.dataflow_options.bq_ops_dataset)
             | "Upload - GA measurement protocol"
             >> beam.ParDo(GoogleAnalyticsMeasurementProtocolUploaderDoFn())
             | "Persist results - GA measurement protocol"
             >> beam.ParDo(
-                TransactionalEventsResultsWriter(self.params._dataflow_options.bq_ops_dataset)
+                TransactionalEventsResultsWriter(
+                  self.params._dataflow_options.bq_ops_dataset,
+                  TransactionalType.UUID)
             )
         )
 
@@ -219,13 +231,15 @@ def expand(self, executions):
             >> BatchesFromExecutions(
                 DestinationType.GA_4_MEASUREMENT_PROTOCOL,
                 20,
-                True,
+                TransactionalType.UUID,
                 self.params.dataflow_options.bq_ops_dataset)
             | "Upload - GA 4 measurement protocol"
             >> beam.ParDo(GoogleAnalytics4MeasurementProtocolUploaderDoFn())
             | "Persist results - GA 4 measurement protocol"
             >> beam.ParDo(
-                TransactionalEventsResultsWriter(self.params._dataflow_options.bq_ops_dataset)
+                TransactionalEventsResultsWriter(
+                  self.params._dataflow_options.bq_ops_dataset,
+                  TransactionalType.UUID)
             )
         )
 
@@ -238,15 +252,17 @@ def expand(self, executions):
             >> BatchesFromExecutions(
                 DestinationType.CM_OFFLINE_CONVERSION,
                 1000,
-                True,
+                TransactionalType.UUID,
                 self.params.dataflow_options.bq_ops_dataset)
             | "Upload - CM conversion"
             >> beam.ParDo(
                 CampaignManagerConversionUploaderDoFn(self.params._oauth_credentials)
             )
             | "Persist results - CM conversion"
             >> beam.ParDo(
-                TransactionalEventsResultsWriter(self.params._dataflow_options.bq_ops_dataset)
+                TransactionalEventsResultsWriter(
+                  self.params._dataflow_options.bq_ops_dataset,
+                  TransactionalType.UUID)
             )
         )
 

megalista_dataflow/requirements-no-deps.txt

@@ -0,0 +1 @@
+apache-beam[gcp]==2.36.0

megalista_dataflow/requirements.txt

@@ -1,23 +1,60 @@
-google-ads==12.0.0
-httplib2==0.17.4
-protobuf==3.13.0
-google-api-python-client==1.12.8
-google-cloud-core==1.4.1
-google-cloud-bigquery==1.27.2
-apache-beam[gcp]==2.36.0
-apache-beam==2.36.0
-google-cloud-datastore==1.13.1
-google-apitools==0.5.31
+google-ads==15.0.0
+google-api-python-client==2.37.0
+google-cloud-bigquery==2.34.0
+google-cloud-firestore==2.3.4
+google-cloud-storage==2.1.0
+
+aiohttp==3.6.2
+
+# Test deps
+mypy==0.790
 pytest==5.4.3
 pytest-cov==2.10.0
 pytest-mock==3.2.0
+pytz==2021.3
 requests-mock==1.8.0
-pytz==2020.1
-wheel==0.34.2
-pyarrow==0.17.1
-aiohttp==3.6.2
-bloom-filter==1.3
-six==1.13.0
-mypy==0.790
-google-cloud-storage==1.38.0
-google-cloud-firestore==2.1.1
+
+
+
+#apache beam deps below
+google-apitools==0.5.31
+cachetools==4.2.4
+certifi==2021.10.8
+charset-normalizer==2.0.12
+cloudpickle==2.0.0
+crcmod==1.7
+dill==0.3.1.1
+docopt==0.6.2
+fastavro==1.4.9
+fasteners==0.17.3
+google-resumable-media==2.2.1
+googleapis-common-protos==1.54.0
+grpc-google-iam-v1==0.12.3
+grpcio==1.44.0
+grpcio-gcp==0.2.2
+grpcio-status==1.44.0
+hdfs==2.6.0
+httplib2==0.19.1
+idna==3.3
+libcst==0.4.1
+mypy-extensions==0.4.3
+numpy==1.21.5
+oauth2client==4.1.3
+orjson==3.6.7
+overrides==6.1.0
+packaging==21.3
+pyarrow==6.0.1
+pyasn1==0.4.8
+pyasn1-modules==0.2.8
+pydot==1.4.2
+pymongo==3.12.3
+pyparsing==2.4.7
+python-dateutil==2.8.2
+# pyyaml==6.0
+requests==2.27.1
+rsa==4.8
+six==1.16.0
+typing-extensions==4.1.1
+typing-inspect==0.7.1
+typing-utils==0.1.0
+urllib3==1.26.8

megalista_dataflow/setup.py

@@ -20,9 +20,8 @@
     author='Google',
     author_email='[email protected]',
     url='https://github.com/google/megalista/',
-    install_requires=['google-ads==12.0.0', 'google-api-python-client==1.12.8',
-                      'google-cloud-core==1.4.1', 'google-cloud-bigquery==1.27.2',
-                      'google-cloud-datastore==1.13.1', 'aiohttp==3.6.2',
-                      'google-cloud-storage==1.38.0', 'google-cloud-firestore==2.1.1'],
+    install_requires=['google-ads==15.0.0', 'google-api-python-client==2.37.0',
+                      'google-cloud-bigquery==2.34.0','aiohttp==3.6.2',
+                      'google-cloud-storage==2.1.0', 'google-cloud-firestore==2.3.4'],
     packages=setuptools.find_packages(),
 )

megalista_dataflow/sources/batches_from_executions.py

@@ -11,6 +11,7 @@
 # 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.
+from enum import Enum
 
 import apache_beam as beam
 import logging
@@ -20,6 +21,7 @@
 from apache_beam.options.value_provider import ValueProvider
 from google.cloud import bigquery
 from models.execution import DestinationType, Execution, Batch
+from string import Template
 from typing import Any, List, Iterable, Tuple, Dict
 
 
@@ -48,6 +50,20 @@ def is_deterministic(self):
         return True
 
 
+class TransactionalType(Enum):
+    """
+        Distinct types to handle data uploading deduplication.
+        NOT_TRANSACTION: don't handle.
+        UUID: Expect a 'uuid' field in the source table as a unique identifier to each row.
+        GCLID_DATE_TIME: Expect 'gclid' and 'time' fields in the source table as unique identifiers to each row.
+    """
+    (
+        NOT_TRANSACTIONAL,
+        UUID,
+        GCLID_TIME,
+    ) = range(3)
+
+
 class BatchesFromExecutions(beam.PTransform):
     """
     Filter the received executions by the received action,
@@ -62,14 +78,15 @@ def process(self, execution: Execution) -> Iterable[Tuple[Execution, Dict[str, A
             table_name = table_name.replace('`', '')
             query = f"SELECT data.* FROM `{table_name}` AS data"
             logging.getLogger(_LOGGER_NAME).info(f'Reading from table {table_name} for Execution {execution}')
-            rows_iterator = client.query(query).result(page_size=_BIGQUERY_PAGE_SIZE)
-            for row in rows_iterator:
+            for row in client.query(query).result(page_size=_BIGQUERY_PAGE_SIZE):
                 yield execution, _convert_row_to_dict(row)
 
     class _ExecutionIntoBigQueryRequestTransactional(beam.DoFn):
 
-        def __init__(self, bq_ops_dataset):
+        def __init__(self, bq_ops_dataset, create_table_query, join_query):
             self._bq_ops_dataset = bq_ops_dataset
+            self._create_table_query = create_table_query
+            self._join_query = join_query
 
         def process(self, execution: Execution) -> Iterable[Tuple[Execution, Dict[str, Any]]]:
             table_name = execution.source.source_metadata[0] + \
@@ -81,25 +98,20 @@ def process(self, execution: Execution) -> Iterable[Tuple[Execution, Dict[str, A
             uploaded_table_name = uploaded_table_name.replace('`', '')
             client = bigquery.Client()
 
-            query = f"CREATE TABLE IF NOT EXISTS `{uploaded_table_name}` ( \
-              timestamp TIMESTAMP OPTIONS(description= 'Event timestamp'), \
-              uuid STRING OPTIONS(description='Event unique identifier')) \
-              PARTITION BY _PARTITIONDATE \
-              OPTIONS(partition_expiration_days=15)"
+            create_table_query_ready = \
+                Template(self._create_table_query).substitute(uploaded_table_name=uploaded_table_name)
 
             logging.getLogger(_LOGGER_NAME).info(
                 f"Creating table {uploaded_table_name} if it doesn't exist")
 
-            client.query(query).result()
+            client.query(create_table_query_ready).result()
 
-            query = f"SELECT data.* FROM `{table_name}` AS data \
-                LEFT JOIN {uploaded_table_name} AS uploaded USING(uuid) \
-                WHERE uploaded.uuid IS NULL;"
+            join_query_ready = \
+                Template(self._join_query).substitute(table_name=table_name, uploaded_table_name=uploaded_table_name)
 
             logging.getLogger(_LOGGER_NAME).info(
                 f'Reading from table {table_name} for Execution {execution}')
-            rows_iterator = client.query(query).result(page_size=_BIGQUERY_PAGE_SIZE)
-            for row in rows_iterator:
+            for row in client.query(join_query_ready).result(page_size=_BIGQUERY_PAGE_SIZE):
                 yield execution, _convert_row_to_dict(row)
 
 
@@ -123,21 +135,44 @@ def __init__(
         self,
         destination_type: DestinationType,
         batch_size: int = 5000,
-        transactional: bool = False,
+        transactional_type: TransactionalType = TransactionalType.NOT_TRANSACTIONAL,
         bq_ops_dataset: ValueProvider = None
     ):
         super().__init__()
-        if transactional and not bq_ops_dataset:
+        if transactional_type is not TransactionalType.NOT_TRANSACTIONAL and not bq_ops_dataset:
             raise Exception('Missing bq_ops_dataset for this uploader')
 
         self._destination_type = destination_type
         self._batch_size = batch_size
-        self._transactional = transactional
+        self._transactional_type = transactional_type
         self._bq_ops_dataset = bq_ops_dataset
 
     def _get_bq_request_class(self):
-        if self._transactional:
-            return self._ExecutionIntoBigQueryRequestTransactional(self._bq_ops_dataset)
+        if self._transactional_type == TransactionalType.UUID:
+            return self._ExecutionIntoBigQueryRequestTransactional(
+                self._bq_ops_dataset,
+                "CREATE TABLE IF NOT EXISTS `$uploaded_table_name` ( \
+                             timestamp TIMESTAMP OPTIONS(description= 'Event timestamp'), \
+                             uuid STRING OPTIONS(description='Event unique identifier')) \
+                             PARTITION BY _PARTITIONDATE \
+                             OPTIONS(partition_expiration_days=15)",
+                "SELECT data.* FROM `$table_name` AS data \
+                               LEFT JOIN $uploaded_table_name AS uploaded USING(uuid) \
+                               WHERE uploaded.uuid IS NULL;"
+                )
+        if self._transactional_type == TransactionalType.GCLID_TIME:
+            return self._ExecutionIntoBigQueryRequestTransactional(
+                self._bq_ops_dataset,
+                "CREATE TABLE IF NOT EXISTS `$uploaded_table_name` ( \
+                             timestamp TIMESTAMP OPTIONS(description= 'Event timestamp'), \
+                             gclid STRING OPTIONS(description= 'Original gclid'), \
+                             time STRING OPTIONS(description= 'Original time')) \
+                             PARTITION BY _PARTITIONDATE \
+                             OPTIONS(partition_expiration_days=15)",
+                "SELECT data.* FROM `$table_name` AS data \
+                               LEFT JOIN $uploaded_table_name AS uploaded USING(gclid, time) \
+                               WHERE uploaded.gclid IS NULL;"
+            )
         return self._ExecutionIntoBigQueryRequest()
 
     def expand(self, executions):

megalista_dataflow/third_party/steps.py

@@ -7,7 +7,8 @@
 from third_party.uploaders.appsflyer.appsflyer_s2s_uploader_async import AppsFlyerS2SUploaderDoFn
 from models.execution import DestinationType
 from uploaders.big_query.transactional_events_results_writer import TransactionalEventsResultsWriter
-from sources.batches_from_executions import BatchesFromExecutions
+from sources.batches_from_executions import BatchesFromExecutions, TransactionalType
+
 
 class AppsFlyerEventsStep(beam.PTransform):
     def __init__(self, params):
@@ -20,9 +21,12 @@ def expand(self, executions):
             BatchesFromExecutions(
                 DestinationType.APPSFLYER_S2S_EVENTS,
                 1000,
-                True,
+                TransactionalType.UUID,
                 self.params.dataflow_options.bq_ops_dataset)
             | 'Upload - AppsFlyer S2S events' >>
             beam.ParDo(AppsFlyerS2SUploaderDoFn(self.params.dataflow_options.appsflyer_dev_key))
-            | 'Persist results - AppsFlyer S2S events' >> beam.ParDo(TransactionalEventsResultsWriter(self.params.dataflow_options.bq_ops_dataset))
+            | 'Persist results - AppsFlyer S2S events' >> beam.ParDo(
+                            TransactionalEventsResultsWriter(
+                              self.params.dataflow_options.bq_ops_dataset,
+                              TransactionalType.UUID))
         )