Facebook Marketing performance improvement

Author: Dmytro Rezchykov

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/api.py

@@ -23,6 +23,17 @@ class MyFacebookAdsApi(FacebookAdsApi):
     call_rate_threshold = 90  # maximum percentage of call limit utilization
     pause_interval_minimum = pendulum.duration(minutes=1)  # default pause interval if reached or close to call rate limit
 
+    # from 0 to 1.0
+    _ads_insights_throttle: float = None
+
+    @property
+    def ads_insights_throttle(self):
+        return self._ads_insights_throttle
+
+    @ads_insights_throttle.setter
+    def ads_insights_throttle(self, value):
+        self._ads_insights_throttle = value
+
     @staticmethod
     def parse_call_rate_header(headers):
         usage = 0
@@ -90,6 +101,10 @@ def call(
     ):
         """Makes an API call, delegate actual work to parent class and handles call rates"""
         response = super().call(method, path, params, headers, files, url_override, api_version)
+        ads_insights_throttle = response.headers().get("x-fb-ads-insights-throttle")
+        if ads_insights_throttle:
+            ads_insights_throttle = json.loads(ads_insights_throttle)
+            self.ads_insights_throttle = max(ads_insights_throttle.get("app_id_util_pct"), ads_insights_throttle.get("acc_id_util_pct"))
         self.handle_call_rate_limit(response, params)
         return response
 

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/async_job.py

@@ -8,30 +8,29 @@
 
 import backoff
 import pendulum
+from facebook_business.adobjects.adreportrun import AdReportRun
 from facebook_business.exceptions import FacebookRequestError
 from source_facebook_marketing.api import API
 
-from .common import JobException, JobTimeoutException, retry_pattern
+from .common import JobException, retry_pattern
 
 backoff_policy = retry_pattern(backoff.expo, FacebookRequestError, max_tries=5, factor=5)
 logger = logging.getLogger("airbyte")
 
 
-class Status(Enum):
+class Status(str, Enum):
     """Async job statuses"""
 
     COMPLETED = "Job Completed"
     FAILED = "Job Failed"
     SKIPPED = "Job Skipped"
     STARTED = "Job Started"
+    NOT_STARTED = "Job Not Started"
 
 
 class AsyncJob:
     """AsyncJob wraps FB AdReport class and provides interface to restart/retry the async job"""
 
-    MAX_WAIT_TO_START = pendulum.duration(minutes=5)
-    MAX_WAIT_TO_FINISH = pendulum.duration(minutes=30)
-
     def __init__(self, api: API, params: Mapping[str, Any]):
         """Initialize
 
@@ -40,7 +39,7 @@ def __init__(self, api: API, params: Mapping[str, Any]):
         """
         self._params = params
         self._api = api
-        self._job = None
+        self._job: AdReportRun = None
         self._start_time = None
         self._finish_time = None
         self._failed = False
@@ -90,7 +89,7 @@ def completed(self) -> bool:
             return self._check_status()
         except JobException:
             self._failed = True
-            raise
+            return True
 
     @property
     def failed(self) -> bool:
@@ -113,25 +112,14 @@ def _check_status(self) -> bool:
         job_progress_pct = self._job["async_percent_completion"]
         logger.info(f"{self} is {job_progress_pct}% complete ({self._job['async_status']})")
         runtime = self.elapsed_time
+        job_status = Status(self._job["async_status"])
 
-        if self._job["async_status"] == Status.COMPLETED.value:
+        if job_status == Status.COMPLETED:
             self._finish_time = pendulum.now()
             return True
-        elif self._job["async_status"] == Status.FAILED.value:
-            raise JobException(f"{self._job} failed after {runtime.in_seconds()} seconds.")
-        elif self._job["async_status"] == Status.SKIPPED.value:
-            raise JobException(f"{self._job} skipped after {runtime.in_seconds()} seconds.")
-
-        if runtime > self.MAX_WAIT_TO_START and self._job["async_percent_completion"] == 0:
-            raise JobTimeoutException(
-                f"{self._job} did not start after {runtime.in_seconds()} seconds."
-                f" This is an intermittent error which may be fixed by retrying the job. Aborting."
-            )
-        elif runtime > self.MAX_WAIT_TO_FINISH:
-            raise JobTimeoutException(
-                f"{self._job} did not finish after {runtime.in_seconds()} seconds."
-                f" This is an intermittent error which may be fixed by retrying the job. Aborting."
-            )
+        elif job_status in [Status.FAILED, Status.SKIPPED]:
+            raise JobException(f"{self._job} has status {job_status} after {runtime.in_seconds()} seconds.")
+
         return False
 
     @backoff_policy

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/async_job_manager.py

@@ -0,0 +1,84 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import logging
+import time
+from collections import deque
+from dataclasses import dataclass
+
+import pendulum
+from source_facebook_marketing.api import API
+
+from .async_job import AsyncJob
+
+
+@dataclass
+class InsightsAsyncJobManager:
+
+    logger = logging.getLogger("airbyte")
+    api: API
+    from_date: pendulum.Date
+    to_date: pendulum.Date
+    start_days_per_job: int
+    job_params: dict
+    _jobs_queue: deque = deque()
+
+    THROTTLE_LIMIT = 0.7
+    FAILED_JOBS_RESTART_COUNT = 5
+    JOB_STATUS_UPDATE_SLEEP_SECONDS = 30
+
+    def done(self) -> bool:
+        """docstring for done"""
+        return len(self._jobs_queue) == 0
+
+    def _current_throttle(self):
+        return self.api.api.ads_insights_throttle
+
+    def get_next_range(self):
+        until = min(
+            self.from_date + pendulum.Duration(days=self.start_days_per_job),
+            self.to_date,
+        )
+        try:
+            return {
+                "time_range": {
+                    "since": self.from_date.to_date_string(),
+                    "until": until.to_date_string(),
+                },
+            }
+        finally:
+            self.from_date = until.add(days=1)
+
+    def no_more_ranges(self) -> bool:
+        return self.from_date >= self.to_date
+
+    def add_async_jobs(self):
+        if self.no_more_ranges():
+            return
+        self._update_api_throttle_limit()
+        while self._current_throttle() < self.THROTTLE_LIMIT and not self.no_more_ranges():
+            next_range = self.get_next_range()
+            params = {**self.job_params, **next_range}
+            job = AsyncJob(api=self.api, params=params)
+            job.start()
+            self._jobs_queue.append(job)
+        self.logger.info(f"Current throttle limit is {self._current_throttle()}")
+
+    def get_next_completed_job(self) -> AsyncJob:
+        job = self._jobs_queue[0]
+        for _ in range(self.FAILED_JOBS_RESTART_COUNT):
+            while not job.completed:
+                self.logger.info(f"Job {job} is not ready, wait for {self.JOB_STATUS_UPDATE_SLEEP_SECONDS}")
+                time.sleep(self.JOB_STATUS_UPDATE_SLEEP_SECONDS)
+            if job.failed:
+                self.logger(f"Job {job} failed, restarting")
+                job.restart()
+            self.add_async_jobs()
+            return self._jobs_queue.popleft()
+        else:
+            # TODO: Break range into smaller parts if job failing constantly
+            raise Exception(f"{job} failed")
+
+    def _update_api_throttle_limit(self):
+        self.api.account.get_insights()

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/streams.py

@@ -2,10 +2,8 @@
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
 
-import time
 import urllib.parse as urlparse
 from abc import ABC
-from collections import deque
 from datetime import datetime
 from typing import Any, Iterable, Iterator, List, Mapping, MutableMapping, Optional, Sequence
 
@@ -21,9 +19,9 @@
 from facebook_business.api import FacebookAdsApiBatch, FacebookRequest, FacebookResponse
 from facebook_business.exceptions import FacebookRequestError
 from source_facebook_marketing.api import API
+from source_facebook_marketing.async_job_manager import InsightsAsyncJobManager
 
-from .async_job import AsyncJob
-from .common import FacebookAPIException, JobException, batch, deep_merge, retry_pattern
+from .common import FacebookAPIException, batch, deep_merge, retry_pattern
 
 backoff_policy = retry_pattern(backoff.expo, FacebookRequestError, max_tries=5, factor=5)
 
@@ -291,8 +289,8 @@ class AdsInsights(FBMarketingIncrementalStream):
     ]
 
     MAX_ASYNC_SLEEP = pendulum.duration(minutes=5)
-    MAX_ASYNC_JOBS = 10
-    INSIGHTS_RETENTION_PERIOD = pendulum.duration(days=37 * 30)
+    MAX_ASYNC_JOBS = 1000
+    INSIGHTS_RETENTION_PERIOD_MONTHES = 37
 
     action_breakdowns = ALL_ACTION_BREAKDOWNS
     level = "ad"
@@ -314,7 +312,7 @@ def __init__(
 
         super().__init__(**kwargs)
         self.lookback_window = pendulum.duration(days=buffer_days)
-        self._days_per_job = days_per_job
+        self._days_per_job = 5  # days_per_job
         self._fields = fields
         self.action_breakdowns = action_breakdowns or self.action_breakdowns
         self.breakdowns = breakdowns or self.breakdowns
@@ -336,7 +334,7 @@ def read_records(
         stream_state: Mapping[str, Any] = None,
     ) -> Iterable[Mapping[str, Any]]:
         """Waits for current job to finish (slice) and yield its result"""
-        job = self.wait_for_job(stream_slice["job"])
+        job = stream_slice["job"]
         # because we query `lookback_window` days before actual cursor we might get records older then cursor
 
         for obj in job.get_result():
@@ -349,50 +347,37 @@ def stream_slices(self, stream_state: Mapping[str, Any] = None, **kwargs) -> Ite
         2. we should run as many job as possible before checking for result
         3. we shouldn't proceed to consumption of the next job before previous succeed
         """
-        stream_state = stream_state or {}
-        running_jobs = deque()
-        date_ranges = list(self._date_ranges(stream_state=stream_state))
-        for params in date_ranges:
-            params = deep_merge(params, self.request_params(stream_state=stream_state))
-            job = AsyncJob(api=self._api, params=params)
-            job.start()
-            running_jobs.append(job)
-            if len(running_jobs) >= self.MAX_ASYNC_JOBS:
-                yield {"job": running_jobs.popleft()}
-
-        while running_jobs:
-            yield {"job": running_jobs.popleft()}
-
-    @retry_pattern(backoff.expo, JobException, max_tries=10, factor=5)
-    def wait_for_job(self, job: AsyncJob) -> AsyncJob:
-        if job.failed:
-            job.restart()
-
-        factor = 2
-        sleep_seconds = factor
-        while not job.completed:
-            self.logger.info(f"{job}: sleeping {sleep_seconds} seconds while waiting for completion")
-            time.sleep(sleep_seconds)
-            if sleep_seconds < self.MAX_ASYNC_SLEEP.in_seconds():
-                sleep_seconds *= factor
-
-        return job
 
-    def request_params(self, stream_state: Mapping[str, Any], **kwargs) -> MutableMapping[str, Any]:
-        params = super().request_params(stream_state=stream_state, **kwargs)
-        params = deep_merge(
-            params,
-            {
-                "level": self.level,
-                "action_breakdowns": self.action_breakdowns,
-                "breakdowns": self.breakdowns,
-                "fields": self.fields,
-                "time_increment": self.time_increment,
-                "action_attribution_windows": self.action_attribution_windows,
-            },
+        job_params = self.request_params(stream_state=stream_state)
+        job_manager = InsightsAsyncJobManager(
+            api=self._api,
+            job_params=job_params,
+            start_days_per_job=5,
+            from_date=self.get_start_date(stream_state),
+            to_date=self._end_date,
         )
+        job_manager.add_async_jobs()
 
-        return params
+        while not job_manager.done():
+            yield {"job": job_manager.get_next_completed_job()}
+
+    def get_start_date(self, stream_state: Mapping[str, Any]) -> pendulum.Date:
+        state_value = stream_state.get(self.cursor_field) if stream_state else None
+        if state_value:
+            start_date = pendulum.parse(state_value) - self.lookback_window
+        else:
+            start_date = self._start_date
+        return max(self._end_date.subtract(months=self.INSIGHTS_RETENTION_PERIOD_MONTHES), start_date)
+
+    def request_params(self, stream_state: Mapping[str, Any], **kwargs) -> MutableMapping[str, Any]:
+        return {
+            "level": self.level,
+            "action_breakdowns": self.action_breakdowns,
+            "breakdowns": self.breakdowns,
+            "fields": self.fields,
+            "time_increment": self.time_increment,
+            "action_attribution_windows": self.action_attribution_windows,
+        }
 
     def _state_filter(self, stream_state: Mapping[str, Any]) -> Mapping[str, Any]:
         """Works differently for insights, so remove it"""
@@ -435,26 +420,6 @@ def _schema_for_breakdowns(self) -> Mapping[str, Any]:
 
         return {breakdown: schemas[breakdown] for breakdown in self.breakdowns}
 
-    def _date_ranges(self, stream_state: Mapping[str, Any]) -> Iterator[dict]:
-        """Iterate over period between start_date/state and now
-
-        Notes: Facebook freezes insight data 28 days after it was generated, which means that all data
-            from the past 28 days may have changed since we last emitted it, so we retrieve it again.
-        """
-        state_value = stream_state.get(self.cursor_field)
-        if state_value:
-            start_date = pendulum.parse(state_value) - self.lookback_window
-        else:
-            start_date = self._start_date
-        end_date = self._end_date
-        start_date = max(end_date - self.INSIGHTS_RETENTION_PERIOD, start_date)
-
-        for since in pendulum.period(start_date, end_date).range("days", self._days_per_job):
-            until = min(since.add(days=self._days_per_job - 1), end_date)  # -1 because time_range is inclusive
-            yield {
-                "time_range": {"since": since.to_date_string(), "until": until.to_date_string()},
-            }
-
 
 class AdsInsightsAgeAndGender(AdsInsights):
     breakdowns = ["age", "gender"]