Documentation
Email CampaignsWebsite PersonalizationMobile AppsWeb PushLoyalty Program

How to update data periodically

The data reading algorithm looks like this:

  • create tables in your DWH
  • load historical data from version zero
  • periodically read data from version N+1, where N is the latest version of data in your DWH
  • get links to the necessary files and read the contents of the files
  • put the data from the files into DWH, fixing the version of the data so that in the next iteration of reading, start reading from the version that has not yet been written to your DWH, and not read everything from the beginning

Example of tables and views for Clickhouse

Tables for raw data

Balance changes

CREATE TABLE "BonusPointChanges_raw" (
"id" Int,
"kindSystemName" Nullable(String),
"mechanicsInternalId" Nullable(String),
"balanceInternalId" Nullable(String),
"changeAmount" Decimal64 (5),
"availableFromDateTimeUtc" Nullable(String),
"expirationDateTimeUtc" Nullable(String),
"dateTimeUtc" Nullable(String),
"unmergedCustomerId" Nullable(String),
"orderId" Nullable(String),
"comments" Nullable(String),
"pointOfContactInternalId" Nullable(String),
"brandInternalId" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int
)
ENGINE MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192 SETTINGS flatten_nested=0

Mechanics of points balance changes

CREATE TABLE "BonusPointsMechanics_raw" (
"id" Int,
"internalId" String,
"discriminator" Nullable(String),
"name" Nullable(String),
"ownerId" Nullable(String),
"ownerType" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int
)
ENGINE MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192 SETTINGS flatten_nested=0

Points accounts

CREATE TABLE "Balances_raw" (
"internalId" String,
"id" Nullable(Int),
"name" Nullable(String),
"systemName" Nullable(String),
"description" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int
)
ENGINE MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192 SETTINGS flatten_nested=0

Links between write-offs and accruals

CREATE TABLE "NegativeCustomerBalanceChangeDetails_raw" (
"id" Int,
"negativeCustomerBalanceChangeId" Nullable(Int),
"positiveCustomerBalanceChangeId" Nullable(Int),
"spentAmount" Decimal64 (5),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int
)
ENGINE MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192 SETTINGS flatten_nested=0

Orders

CREATE TABLE "Orders_raw" (
"id" String,
"unmergedCustomerId" Nullable(Int64),
"firstDateTimeUtc" Nullable(DateTime),
"firstPointOfContactInternalId" Nullable(String),
"firstBrandInternalId" Nullable(String),
"price" Nullable(Decimal64 (5)),
"priceWithDiscounts" Nullable(Decimal64 (5)),
"deliveryPrice" Nullable(Decimal64 (5)),
"deliveryPriceWithDiscounts" Nullable(Decimal64 (5)),
"paidAmount" Nullable(Decimal64 (5)),
"pointOfContactInternalId" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int
)
ENGINE MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192 SETTINGS flatten_nested=0

Touchpoints

CREATE TABLE "PointsOfContact_raw" (
"id" Int64,
"internalId" String,
"externalId" Nullable(String),
"name" String,
"systemName" String,
"parentId" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Purchase statuses dictionary

CREATE TABLE "PurchaseStatuses_raw" (
"internalId" String,
"name" Nullable(String),
"externalId" Nullable(String),
"categorySystemName" Nullable(String),
"_isDeleted" Nullable(String),
"_rowversion_ts" DateTime,
"_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Purchases

The purchase record does not have a _isDeleted field because it cannot be deleted - only the entire order can be deleted.

CREATE TABLE "Purchases_raw" (
"orderId" String,
"pricePerItem" Decimal64 (5),
"priceOfLine" Decimal64 (5),
"quantity" Float,
"quantityType" String,
"lineId" Nullable(String),
"lineNumber" Int,
"statusInternalId" String,
"productInternalId" String,
"_rowversion_ts" DateTime,
"_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Mailings dictionary

CREATE TABLE "Mailings_raw" (
    "id" String,
    "name" String,
    "systemName" String,
    "type" String,
    "channel" String,
    "creationDateTimeUtc" DateTime,
    "lastUpdateDateTimeUtc" DateTime,
    "folderInternalId" Nullable(String),
    "subscriptionTopicInternalId" Nullable(String),
    "brandInternalId" String,
    "utmSource" Nullable(String),
    "utmMedium" Nullable(String),
    "utmCampaign" Nullable(String),
    "utmContent" Nullable(String),
    "utmTerm" Nullable(String),
    "_isDeleted" Nullable(String),
    "_rowversion_ts" DateTime,
    "_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Subscription topics dictionary

CREATE TABLE "SubscriptionTopics_raw"
(
    "internalId" String,
    "systemName" String,
    "name" String,
    "brandInternalId" String,
    "_isDeleted" Nullable(String),
    "_rowversion_ts" DateTime,
    "_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Mailings statuses

CREATE TABLE "CustomerMessagesStatuses_raw"
(
    "messageId" Int64,
    "messageStatusId" String,
    "mailingStatusSystemName" String,
    "dateTimeUtc" DateTime,
    "unmergedCustomerId" Int64,
    "mailingInternalId" String,
    "mailingVariantNum" Nullable(String),
    "mailingLink" Nullable(String),
    "mailingSourceEntityType" Nullable(String),
    "mailingSourceEntityId" Nullable(String),
    "notSentSystemName" Nullable(String),
    "notDeliveredReasonSystemName" Nullable(String),
    "_isDeleted" Nullable(String),
    "_rowversion_ts" DateTime,
    "_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Customer merges

CREATE TABLE "MergedCustomers_raw"
(
    "unmergedCustomerId" Int64,
    "mergedCustomerId" Int64,
    "dateTimeUtc" DateTime,
    "_isDeleted" Nullable(String),
    "_rowversion_ts" DateTime,
    "_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Folders

CREATE TABLE "Folders_raw"
(
    "internalId" String,
    "systemName" String,
    "name" String,
    "parentInternalId" Nullable(String),
    "_isDeleted" Nullable(String),
    "_rowversion_ts" DateTime,
    "_data_version" Int32
)
ENGINE = MergeTree()
ORDER BY tuple()
SETTINGS index_granularity = 8192

Datamarts with the current state of data

These views will only contain current data without manually deleted data and without duplicates - only the latest states of facts and entities

Balance changes

CREATE VIEW BonusPointChanges_dm AS
SELECT * FROM ( 
   SELECT * 
   FROM 
   BonusPointChanges_raw
   ORDER BY 
   BonusPointChanges_raw._rowversion_ts DESC
   LIMIT 1 BY BonusPointChanges_raw.id
) AS dm
WHERE 
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Mechanics of points balance changes

CREATE VIEW BonusPointsMechanics_dm AS
SELECT * FROM ( 
   SELECT * 
   FROM 
   BonusPointsMechanics_raw
   ORDER BY 
   BonusPointsMechanics_raw._rowversion_ts DESC
   LIMIT 1 BY BonusPointsMechanics_raw.id
) AS dm
WHERE 
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Points accounts

CREATE VIEW Balances_dm AS
SELECT * FROM ( 
   SELECT * 
   FROM 
   Balances_raw
   ORDER BY 
   Balances_raw._rowversion_ts DESC
   LIMIT 1 BY Balances_raw.id
) AS dm
WHERE 
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Links between write-offs and accruals

CREATE VIEW NegativeCustomerBalanceChangeDetails_dm AS
SELECT * FROM ( 
   SELECT * 
   FROM 
   NegativeCustomerBalanceChangeDetails_raw
   ORDER BY 
   NegativeCustomerBalanceChangeDetails_raw._rowversion_ts DESC
   LIMIT 1 BY NegativeCustomerBalanceChangeDetails_raw.id
) AS dm
WHERE 
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Orders

CREATE VIEW Orders_dm AS
SELECT * FROM ( 
   SELECT * 
   FROM 
   Orders_raw
   ORDER BY 
   Orders_raw._rowversion_ts DESC
   LIMIT 1 BY Orders_raw.id
) AS dm
WHERE 
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Touchpoints

CREATE VIEW PointsOfContact_dm AS
SELECT * FROM (
	SELECT *
	FROM
	PointsOfContact_raw
	ORDER BY
	PointsOfContact_raw._rowversion_ts DESC
	LIMIT 1 BY PointsOfContact_raw.id
) AS dm
WHERE
empty (dm._isDeleted) OR dm._isDeleted = 'false'

Purchase statuses dictionary

CREATE VIEW PurchaseStatuses_dm AS 
SELECT * FROM
(
    SELECT *
    FROM PurchaseStatuses_raw
    ORDER BY PurchaseStatuses_raw._rowversion_ts DESC
    LIMIT 1 BY PurchaseStatuses_raw.internalId
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Purchases

CREATE VIEW Purchases_dm AS
SELECT *
FROM Purchases_raw
ORDER BY Purchases_raw._rowversion_ts DESC
LIMIT 1 BY Purchases_raw.orderId, Purchases_raw.lineId

Mailings dictionary

CREATE VIEW Mailings_dm AS
SELECT *
FROM
(
    SELECT *
    FROM Mailings.Mailings_raw
    ORDER BY Mailings_raw._rowversion_ts DESC
    LIMIT 1 BY Mailings_raw.id
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Subscription topics

CREATE VIEW SubscriptionTopics_dm AS
SELECT *
FROM
(
    SELECT *
    FROM SubscriptionTopics_raw
    ORDER BY SubscriptionTopics_raw._rowversion_ts DESC
    LIMIT 1 BY SubscriptionTopics_raw.internalId
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Mailing statuses

CREATE VIEW CustomerMessagesStatuses_dm AS
SELECT *
FROM
(
    SELECT *
    FROM CustomerMessagesStatuses_raw
    ORDER BY CustomerMessagesStatuses_raw._rowversion_ts DESC
    LIMIT 1 BY CustomerMessagesStatuses_raw.messageStatusId
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Customer merges

CREATE VIEW MergedCustomers_dm AS
SELECT *
FROM
(
    SELECT *
    FROM MergedCustomers_raw
    ORDER BY MergedCustomers_raw._rowversion_ts DESC
    LIMIT 1 BY
        MergedCustomers_raw.unmergedCustomerId,
        MergedCustomers_raw.mergedCustomerId
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Folders

CREATE VIEW Folders_dm AS
SELECT *
FROM
(
    SELECT *
    FROM Folders_raw
    ORDER BY Folders_raw._rowversion_ts DESC
    LIMIT 1 BY Folders_raw.internalId
) AS dm
WHERE empty(dm._isDeleted) OR (dm._isDeleted = 'false')

Example of reading data using python

Using a script below read data and write it to Clickhouse. This script can then be called daily to write updates to Clickhouse.

To work with Clickhouse, the example uses pandahouse library. If necessary, you will need to install this library using pip3 install pandahouse

Before you begin, install the delta_sharing library using pip3 install delta_sharing

import delta_sharing
import pandahouse as ph
from delta_sharing.rest_client import DataSharingRestClient
from delta_sharing.protocol import DeltaSharingProfile
from delta_sharing.rest_client import DataSharingRestClient, HTTPError
from delta_sharing.protocol import CdfOptions
import pathlib
import pyarrow.parquet as pq
import requests
import io
from datetime import datetime
import urllib3
import pandas

### Connection settings for your DWH ###
connection = dict(database='DB in your DWH',
                  host='URL of your DWH',
                  user='Service user login',
                  password='Service user password')

class Tables():          
    def __init__(self, database, schema, source, target):  
        self.database = database # Source DB
        self.schema = schema # Source schema
        self.source = source # Source table
        self.target = target # Destination table in your DWH
        

tables = []
tables.append ( Tables('exports', 'ProcessingOrders', 'BonusPointChanges', 'BonusPointChanges_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'BalanceChangeKinds', 'BalanceChangeKinds_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'BonusPointsMechanics', 'BonusPointsMechanics_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Balances', 'Balances_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'NegativeCustomerBalanceChangeDetails', 'NegativeCustomerBalanceChangeDetails_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Orders', 'Orders_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Purchases', 'Purchases_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'PurchaseStatuses', 'PurchaseStatuses_raw') )
tables.append ( Tables('exports', 'Mailings', 'Mailings', 'Mailings_raw') )
tables.append ( Tables('exports', 'Mailings', 'SubscriptionTopics', 'SubscriptionTopics_raw') )
tables.append ( Tables('exports', 'Mailings', 'CustomerMessagesStatuses', 'CustomerMessagesStatuses_raw') )
tables.append ( Tables('exports', 'CDP', 'MergedCustomers', 'MergedCustomers_raw') )
tables.append ( Tables('exports', 'CDP', 'Folders', 'Folders_raw') )

### Authentication ### 
share_file_path = str(pathlib.Path().resolve()) + "/Profile.json" # Authentication file path

# Batch size for reading - it can be edited to fit into RAM
batch_size = 1_000_000 

# Number of data file versions read at a time 
cdf_batch_size = 10

# Variables fo statistics and logs
total_len = 0 
start_datetime = datetime.now()

# For each table new versions of data are checked and copied to the DWH
for t in tables:
    
    '''Paste your code below to check the last read version of the data that was copied to your DWH'''
    
    # Getting the latest version of data in DVH - we will read everything from DS that has a higher version
    df = ph.read_clickhouse('SELECT _data_version FROM ' + t.target + ' ORDER BY _data_version DESC LIMIT 1', connection=connection)
    
    '''Paste your code above to check the last read version of the data that was copied to your DWH'''
    
    # Checking if there is data in the table, if not, then reading is performed from version 0
    if df.empty:
        latest_version = 0
    else:
        latest_version = df.iat[0,0] + 1
    
    # Reading from the table
    # Creating a client
    profile = DeltaSharingProfile.read_from_file(share_file_path)
    rest_client = DataSharingRestClient(profile)

    table = delta_sharing.Table(share=t.database, schema=t.schema, name=t.source)

    # Getting the latest versions of data in a table - versions that are greater than latest_version
    ver = latest_version

    # Reading in chunks of cdf_batch_size versions at a time due to limit on number of requested versions
    while True: # Reading so far no error that there are no versions available
        i = 0
        while True and i < 5: # 5 read attempts in case the service is unavailable
            i += 1
            try: 
                print (f"Reading data from table {t.source} versions from {ver} to {ver+cdf_batch_size-1}.")
                res = rest_client.list_table_changes(
                    table,
                    cdfOptions=CdfOptions(starting_version=ver, ending_version=ver+cdf_batch_size-1)
                )
            except (delta_sharing.rest_client.HTTPError) as err: # error may occur if no new data is loaded or more than 10 versions are requested at once
                print (f"Did not receive updated data for versions from {ver} to {ver+cdf_batch_size-1} {i} times. Reason: {err.response.text}")
                continue    
            break
        
        # if failed to read the data after 5 attempts, then try to read from the next table
        if i == 5: 
                break
        
        # If new data is received, we write it to Clickhouse
        print (f"Writing data of versions from {ver} to {ver+cdf_batch_size-1} to table {t.target}")
    
        # Reading data in parts by batch_size rows
        for cdcFile in res.actions:
            signed_url = cdcFile.url
                
            def read_parquet_from_signed_url_in_batches(signed_url, batch_size):
                # Getting data file
                response = requests.get(signed_url, stream=True)
                response.raise_for_status()
                buffer = io.BytesIO()

                # Downoading a file
                for chunk in response.iter_content(chunk_size=1024):
                    buffer.write(chunk)

                buffer.seek(0)

                # Opening a file
                parquet_file = pq.ParquetFile(buffer)

                # Reading a file in batches
                for batch in parquet_file.iter_batches(batch_size=batch_size):
                    yield batch

            # Processing parquet file in batches
            for batch in read_parquet_from_signed_url_in_batches(signed_url, batch_size):

                # Convert batch to pandas dataframe
                batch_df = batch.to_pandas()
                    
                # Adding a column with a data version to a dataframe
                batch_df['_data_version'] = cdcFile.version

                if '_tenant' in batch_df.columns: 
                    batch_df = batch_df.drop('_tenant', axis=1) # Removing the column with the project name as it is not needed for analytics

                # Writing to a target table in ClickHouse
                i = 0
                while True and i < 5: # 5 attempts to write - Clickhouse may not respond sometimes
                    i += 1
                    try:
                        '''Paste the code to write the dataframe to your DWH below'''
                        
                        ph.to_clickhouse(batch_df, t.target, index=False, chunksize=batch_size, connection=connection)
                        
                        '''Paste the code to write the dataframe to your DWH above'''
                        
                        # Statistics
                        current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                        message = f"{len(batch_df)} rows written in {i} attempts ({current_datetime})"
                        print(message)
                        total_len += len(batch_df)
                    except (urllib3.exceptions.ConnectionError or urllib3.exceptions.HTTPError) as err:
                        print (err.response.text)
                        continue
                    break

        ver += cdf_batch_size # next iteration of reading cdf_batch_size versions

# Output of information about the volume of lines read, the script execution time and the average data writing speed
current_datetime = datetime.now()
print(f"Written rows: {total_len} - in: {(current_datetime - start_datetime).total_seconds()} seconds. Average read and write speed: {total_len / (current_datetime - start_datetime).total_seconds()} rows per second")

How to read changes over a specific period

Using function get_table_version you can also obtain data that was changed over a specific period.

Function get_table_version has 2 parameters:

  • *table url - string in format {share_file_path}#{database}.{schema}.{table}
  • date in format YYYY-MM-DDThh:mm:ssZ - you will get first version after that date
    For example 2024/10/10 11:45PM 138 version of Orders table was uploaded and 2024/10/11 11:45PM 139 version of Orders table was uploaded. If you call get_table_version with parameters {share_file_path}#exports.ProcessingOrders.Orders and 2024-10-11T00:00:00Z it will return 139 version - the nearest version after given date.
📘

Second parameter is optional - if you leave it empty, you will get maximum version of the table

Below you could find an example of reading data for specific period

import delta_sharing
import pandahouse as ph
from delta_sharing.rest_client import DataSharingRestClient
from delta_sharing.protocol import DeltaSharingProfile
from delta_sharing.rest_client import DataSharingRestClient, HTTPError
from delta_sharing.protocol import CdfOptions
import pathlib
import pyarrow.parquet as pq
import pyarrow as pa
import requests
import io
from datetime import datetime
import urllib3
import pandas
import numpy as np

### Connection settings for your DWH ###
connection = dict(database='DB in your DWH',
                  host='URL of your DWH',
                  user='Service user login',
                  password='Service user password')

# Period boundary dates
date_start = '2024-10-29T00:00:00Z'
date_end = '2024-10-30T00:00:00Z'

class Tables():          
    def __init__(self, database, schema, source, target):  
        self.database = database # Source DB
        self.schema = schema # Source schema
        self.source = source # Source table
        self.target = target # Destination table in your DWH

tables = []
tables.append ( Tables('exports', 'ProcessingOrders', 'Orders', 'Orders_raw') )

### Authentication ### 
share_file_path = str(pathlib.Path().resolve()) + "/Profile.json" # Authentication file path

# Batch size for reading - it can be edited to fit into RAM
batch_size = 1_000_000 

# Number of data file versions read at a time 
cdf_batch_size = 1

# Variables fo statistics and logs
total_len = 0 
start_datetime = datetime.now()

# For each table versions of data over specific period are copied to the DWH
for t in tables:
    print (f"Reading from table {t.source}")
    # Reading from the table
    # Creating a client
    table_url = f"{share_file_path}#{t.database}.{t.schema}.{t.source}"
    profile = DeltaSharingProfile.read_from_file(share_file_path)
    rest_client = DataSharingRestClient(profile)
    table = delta_sharing.Table(share=t.database, schema=t.schema, name=t.source)

    # Finding a date that matches the start of a period
    version_start = delta_sharing.get_table_version (table_url, date_start)
    print (f"Version corresponding to the date {date_start} for a table {t.source} is {version_start}")
    
    # Finding a date that matches the end of a period
    try: 
        version_end = delta_sharing.get_table_version (table_url, date_end) 
    except (delta_sharing.rest_client.HTTPError) as err: # If a date is after the latest data in a table - we will take the maximum version
        print (f"Date {date_end} is after the latest date in a table - getting the maximum version")
        version_end = delta_sharing.get_table_version (table_url) 

    print (f"Version corresponding to the date {date_end} for a table {t.source} is {version_end}")
    

    ver = version_start

    while ver <= version_end: #
        try: 
            print (f"Reading data from table {t.source} versions from {ver} to {ver+cdf_batch_size-1}.")
            res = rest_client.list_table_changes(
                table,
                cdfOptions=CdfOptions(starting_version=ver, ending_version=ver+cdf_batch_size-1)
            )
            print(res)
        except (delta_sharing.rest_client.HTTPError) as err: 
            print (f"Did not receive updated data for versions from {ver} to {ver+cdf_batch_size-1} {i} times. Reason: {err.response.text}")
            break

        
        # If new data is received, we write it to Clickhouse
        print (f"Writing data of versions from {ver} to {ver+cdf_batch_size-1} to table {t.target}")
    
        # Reading data in parts by batch_size rows
        for cdcFile in res.actions:
            signed_url = cdcFile.url
                
            def read_parquet_from_signed_url_in_batches(signed_url, batch_size):
                # Getting data file
                response = requests.get(signed_url, stream=True)
                response.raise_for_status()
                buffer = io.BytesIO()

                # Downoading a file
                for chunk in response.iter_content(chunk_size=1024):
                    buffer.write(chunk)

                buffer.seek(0)

                # Opening a file
                parquet_file = pq.ParquetFile(buffer)

                # Reading a file in batches
                for batch in parquet_file.iter_batches(batch_size=batch_size):
                    yield batch

            # Processing parquet file in batches
            for batch in read_parquet_from_signed_url_in_batches(signed_url, batch_size):

                # Convert batch to pandas dataframe
                batch_df = batch.to_pandas()
                    
                # Adding a column with a data version to a dataframe
                batch_df['_data_version'] = cdcFile.version

                if '_tenant' in batch_df.columns: 
                    batch_df = batch_df.drop('_tenant', axis=1) # Removing the column with the project name as it is not needed for analytics

                # Writing to a target table in ClickHouse
                i = 0
                while True and i < 5: # 5 attempts to write - Clickhouse may not respond sometimes
                    i += 1
                    try:
                        '''Paste the code to write the dataframe to your DWH below'''
                        
                        ph.to_clickhouse(batch_df, t.target, index=False, chunksize=batch_size, connection=connection)
                        
                        '''Paste the code to write the dataframe to your DWH above'''
                        
                        # Statistics
                        current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                        message = f"{len(batch_df)} rows written in {i} attempts ({current_datetime})"
                        print(message)
                        total_len += len(batch_df)
                    except (urllib3.exceptions.ConnectionError or urllib3.exceptions.HTTPError) as err:
                        print (err.response.text)
                        continue
                    break

        ver += cdf_batch_size # next iteration of reading cdf_batch_size versions

# Output of information about the volume of lines read, the script execution time and the average data writing speed
current_datetime = datetime.now()
print(f"Written rows: {total_len} - in: {(current_datetime - start_datetime).total_seconds()} seconds. Average read and write speed: {total_len / (current_datetime - start_datetime).total_seconds()} rows per second")

Example of reading data using spark

Before you begin, install the delta_sharing library using pip3 install delta_sharing

📘

Version of connector

The example uses the latest version of the connector for Apache Spark 3.2.0 at the time of writing. If you have problems with the version, you can check the current one here

import delta_sharing
from pyspark.sql import SparkSession
import pathlib
import delta_sharing
import pyarrow.parquet as pq
import requests
import io
from datetime import datetime
import urllib3

class Tables():          
    def __init__(self, database, schema, source, target):  
        self.database = database # Source DB
        self.schema = schema # Source schema
        self.source = source # Source table
        self.target = target # Destination table in your DWH
        
tables = []
tables.append ( Tables('exports', 'ProcessingOrders', 'BonusPointChanges', 'BonusPointChanges_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'BalanceChangeKinds', 'BalanceChangeKinds_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'BonusPointsMechanics', 'BonusPointsMechanics_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Balances', 'Balances_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'NegativeCustomerBalanceChangeDetails', 'NegativeCustomerBalanceChangeDetails_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Orders', 'Orders_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'Purchases', 'Purchases_raw') )
tables.append ( Tables('exports', 'ProcessingOrders', 'PurchaseStatuses', 'PurchaseStatuses_raw') )
tables.append ( Tables('exports', 'Mailings', 'Mailings', 'Mailings_raw') )
tables.append ( Tables('exports', 'Mailings', 'SubscriptionTopics', 'SubscriptionTopics_raw') )
tables.append ( Tables('exports', 'Mailings', 'CustomerMessagesStatuses', 'CustomerMessagesStatuses_raw') )
tables.append ( Tables('exports', 'CDP', 'MergedCustomers', 'MergedCustomers_raw') )
tables.append ( Tables('exports', 'CDP', 'Folders', 'Folders_raw') )

# Creating SparkSession
spark = (
    SparkSession.builder.config(
        "spark.jars.packages",
        "org.apache.hadoop:hadoop-azure:3.3.1,io.delta:delta-core_2.12:2.2.0,io.delta:delta-sharing-spark_2.12:3.2.0",
    )
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)

### Authentication ### 
share_file_path = str(pathlib.Path().resolve()) + "/Profile.json" # Authentication file path

# Number of data file versions read at a time 
cdf_batch_size = 10

for t in tables:
    '''Paste your code below to check the last read version of the data that was copied to your DWH'''
    
    latest_version = #Place here script for obtaining latest version of data in your dwh
    
    '''Paste your code above to check the last read version of the data that was copied to your DWH'''
        
    table_url = share_file_path + f"#{t.database}.{t.schema}.{t.source}"

    ver = latest_version

    final_version = delta_sharing.get_table_version (table_url) # getting the maximum version in the table

    # Reading in chunks of cdf_batch_size versions at a time due to limit on number of requested versions
    while ver <= final_version: # Reading till the maximum version
        i = 0
        while True and i < 5: # 5 read attempts in case the service is unavailable
            i += 1
            try: 
                print (f"Reading data from table {t.source} versions from {ver} to {ver+cdf_batch_size-1}.")
                res = (
                    spark.read.format("deltaSharing")
                    .option("readChangeFeed", "true")
                    .option("startingVersion", ver)
                    .option("endingVersion", ver+cdf_batch_size-1)
                    .load(table_url)
                )
              except (delta_sharing.rest_client.HTTPError) as err: # The request may fail if there is no new data available or if more than 10 versions are requested in a single call.

                print (f"Did not receive updated data for versions from {ver} to {ver+cdf_batch_size-1} {i} times. Reason: {err.response.text}")
                continue    
            break
        
        # if failed to read the data after 5 attempts, then try to read from the next table
        if i == 5: 
                break      
        
        '''Paste the code to write the dataframe to your DWH below'''

        print (f"Writing data from version from {ver} to {ver+cdf_batch_size-1} into table {t.target}")               
                       
        '''Paste the code to write the dataframe to your DWH above'''

        ver += cdf_batch_size #next iteration

What are the rowversion_ts, isDeleted, _data_version fields?

_rowversion_ts is a timestamp when the data was changed. For example, the upload may include 2 changes to the order status: first it was placed, then paid for. To select the current order status, we need to select the record with the maximum _rowversion_ts value

_isDeleted is a flag that the record was manually deleted from the system if its value is true. Auto-deleted records will be exported like regular data without the _isDeleted = true flag

_data_version is the version of the uploaded data, the value of this field will be increased by 1 with each new upload. We use it to check if there is new data on the server.

Updated 4 months ago

What’s Next