SDK - Questions around Global Config

Hello,

We tried the approaches suggested and found out that there is not much difference in performance for fetching data from CDF.

For capturing the numbers, we used the following data model configuration:

• Project: slb-test
• Model: QA-PERF-DATA (QA_SPACE)
• Version: 1
• View: TimeseriesData
• Properties: externalId, timeseriesId, propertyName, entity, frequency
• Time Range: 15-11-2023 00:00 to 17-11-2023 23:59
• Filters:
• propertyName: TARGET-WATER-INJECTION-RATE              , YEP-GAS-PRODUCTION-RATE, METER-OPENING-GAS-VOLUME    ,UC-CONTRACTOR            ,SAND-PRODUCTION    ,UNIT-INJECTION-COST   ,WATER-PRODUCTION-VOLUME,GAS-SALES-VOLUME, GAS-PRODUCTION-RATE,DIFFERENTIAL-PRESSURE-HOLDING-TRIGGERED-VALUE
• frequency: 15M

The time taken by the code which uses API call for fetching data from CDF is 25.945018529891968 seconds

Details about both the approaches tried and their results are as follows:

Approach 1: Setting max_workers in Global Configuration

Code change:

   global_config.default_client_config = ClientConfig(         credentials=OAuthClientCredentials(token_url=token_url,                                        client_id=client_id,client_secret=client_secret,                                  scopes=scopes),         project=cognite_project,         base_url=f"https://{cognite_cluster}.cognitedata.com",         client_name=client_name,         debug=False,     )     global_config.max_workers = 1     #client_config.max_workers=1     global_config.max_connection_pool_size= 10

    return global_config.default_client_config

Here we return the default_client_config object to the python files and each of them generates client object by passing it to CogniteClient

The performance numbers captured are:

Approach 2: Setting max_workers directly in the ClientConfig:

Code change:

   client_config = ClientConfig(          credentials=OAuthClientCredentials(token_url=token_url,                                   client_id=client_id,client_secret=client_secret,                                   scopes=scopes),         project=cognite_project,         base_url=f"https://{cognite_cluster}.cognitedata.com",         client_name=client_name,         debug=False,     )

    client_config.max_workers=10     global_config.max_connection_pool_size= 32

    return client_config

Here we return the client_config object to the python files and each of them generates client object by passing it to CogniteClient

Performance numbers are:

So, we can see that there is no effect of increasing the max_workers on the time duration of fetching the data.

Could you please let us know how things could be done differently to get the job done in a lesser amount of time?

Hi @Everton Colling sure. Sharing the code for fetching data.

        if aggregates:

            dps = client.time_series.data.retrieve(external_id=external_id,                    start=int(startParam), end=int(endParam),

                  aggregates=aggregates, granularity=granularity,limit=limit)

        elif is_latestvalue:

            dps =                                        client.time_series.data.retrieve_latest(external_id=external_id)

        else:

            dps = client.time_series.data.retrieve(external_id=external_id,                    start=int(startParam), end=int(endParam),limit=limit)                 

Time for some actual benchmark code!

It is surprisingly hard to do right, as we use a singleton pattern in the SDK for the thread pool executor that executes all the parallel calls. Hence, in order to resize the number of threads it has available, we need to explicitly re-create it. This probably explains why in your previous tests you saw little to no effect after altering various configuration options.

The code below assumes a function exists that can provide you with a CogniteClient, - replace that with your own. For the tests, I use a previously created time series that has 5 million datapoints randomly distributed in time between 1970 and now.

Note: It also assumes that you are testing performance for raw, (numeric) datapoints i.e. no aggregates. For aggregates, you need to decide how to count as fetching the 10 different aggregates is not that much slower than just fetching a single aggregate.

import time
from timeit import default_timer as timer

import cognite.client.utils._concurrency as conc
from cognite.client.data_classes import Datapoints, DatapointsArray

client = setup_cog_client("forge-sandbox", "greenfield")
ts = client.time_series.retrieve(
    external_id="benchmark:9-dense-5mill-rand-dist-#1/1"
)
max_workers = [1, 2, 3, 5, 10]


def benchmark_raw_stats(res, t0, t1, client):
    if isinstance(res, (Datapoints, DatapointsArray)):
        res = [res]
    if any(dps.value is None for dps in res):
        raise ValueError("no raw dps found, did you fetch aggregates?")

    n_dps = sum(map(len, res))
    t = t1 - t0
    print(
        f"Dps/sec={n_dps/t:.0f}, (across {len(res)} ts) ~t: {t:.4f} "
        f"sec, N dps: {n_dps}, max_workers={client.config.max_workers}"
    )

for i, mw in enumerate(max_workers, 1):
    if i == 1:
        # A small warm-up meant for the API to give us peak performance:
        client.time_series.data.retrieve_arrays(id=ts.id, limit=1_000_000)
        time.sleep(10)

    # Overriding max_workers isn't enough, as the SDK reuse the threadpool
    # (singleton pattern), so after the first API call has been made, the
    # theadpool size is fixed. We delete it to force re-init:
    client.config.max_workers = mw
    del conc._THREAD_POOL_EXECUTOR_SINGLETON

    t0 = timer()
    res = client.time_series.data.retrieve_arrays(id=ts.id)
    benchmark_raw_stats(res, t0, timer(), client)

    # Make rate limiting less likely...
    if i != len(max_workers):  # skip after last test
        time.sleep(15)

So, what kind of performance do I get on my Mac laptop on Wi-Fi (5 GHz)?

Dps/sec=299473, (across 1 ts) ~t: 16.696 sec, N dps: 5000000, max_workers=1
Dps/sec=692511, (across 1 ts) ~t: 7.2201 sec, N dps: 5000000, max_workers=2
Dps/sec=961558, (across 1 ts) ~t: 5.1999 sec, N dps: 5000000, max_workers=3
Dps/sec=1557154,(across 1 ts) ~t: 3.2110 sec, N dps: 5000000, max_workers=5
Dps/sec=2086234,(across 1 ts) ~t: 2.3967 sec, N dps: 5000000, max_workers=10

Hi @Håkon V. Treider , @Jason Dressel  and  Anders Daljord Morken,

We tried the above solution for fetching data points and we were able to achieve the expected performance number (around 20 seconds) for 1 max worker.

Can you please explain what you meant by fetching aggregates of numeric data points? Assuming we have around 6000 time-series and half a million data points what should be the max workers to attain a performance of ~20 sec for full fetch?

Hello @Håkon V. Treider ,

We tried implementing the given solution in our code as mentioned above. We have an existing multi-threading logic implemented in our code, when we use the del conc._THREAD_POOL_EXECUTOR_SINGLETON with that, we get the performance numbers as expected. But when we remove the multi-threading logic. The performance is not improving.

We are using concurrent.futures.ThreadPoolExecutor for implementing multi-threading in our own code.

Please suggest a way in which we can use the SDK calls alone with your solution, so that we can get the ideal performance.