<jupyter_start><jupyter_text>Script de funciones agregadas datos Cassandra en cluster multidomain<jupyter_code>!pip install mysql-connector==2.1.7
!pip install pandas
!pip install sqlalchemy
#requiere instalación adicional, consultar https://github.com/PyMySQL/mysqlclient
!pip install mysqlclient
!pip install numpy
!pip install pymysql
import pandas as pd
import numpy as np
import os
import json
import random
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
import time
from pprint import pprint
import psutil
import uuid
from cassandra.query import tuple_factory
from cassandra.query import dict_factory
from cassandra.query import BatchStatement, SimpleStatement
from cassandra.policies import RetryPolicy
#Los resultados de medidas de tiempo en carga por dominios se almacenan en estos objetos.
#Se itera durante 100 iteraciones para sacar medias
#repeticiones
repeats = 10
#Ficheros de salida
resultados_etl_agregadas = '../Results/Cassandra/CassandraAgregadas_test_{}.csv'
def save_results_to_csv(results,file):
    #Guardamos los resultados en csv
    from datetime import datetime
    
    csv_df = pd.DataFrame(results, columns=['Registros', 'Tiempo', 'CPU','Memoria'])
    dia = datetime.now().strftime("%d%m%Y_%H_%M_%S")
    print(file.format(str(dia)))
    csv_df.to_csv(file.format(str(dia)))
from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import WhiteListRoundRobinPolicy, DowngradingConsistencyRetryPolicy
from cassandra.query import tuple_factory
from cassandra import ConsistencyLevel

profile = ExecutionProfile(
    load_balancing_policy=WhiteListRoundRobinPolicy(['127.0.0.1']),
    retry_policy=DowngradingConsistencyRetryPolicy(),
    consistency_level=ConsistencyLevel.ALL,
    serial_consistency_level=ConsistencyLevel.LOCAL_SERIAL,
    request_timeout=3600,
    row_factory=tuple_factory
)
cluster = Cluster(execution_profiles={EXEC_PROFILE_DEFAULT: profile})
session = cluster.connect()
print(session.execute("SELECT release_version FROM system.local").one())

session.execute('USE currentaccountkeyspace')<jupyter_output>('4.0.3',)<jupyter_text>Select test multidomain<jupyter_code>select_query_mysql = """SELECT AVG(Amount),MIN(Amount),MAX(Amount) FROM PositionKeepingDomainSchema.Amount am
INNER JOIN PositionKeepingDomainSchema.PositionKeeping pk ON pk.AmountId = am.AmountId
INNER JOIN CurrentAccountDomainSchema.CurrentAccount ca ON ca.AccountId = pk.AccountId
INNER JOIN CustomerProfileDomainSchema.CustomerProfile cp ON cp.PartyId = ca.PartyId
INNER JOIN CurrentAccountDomainSchema.AccountInfo ai ON ai.AccountId = ca.AccountId
INNER JOIN PositionKeepingDomainSchema.Currency cr ON cr.CurrencyId = am.CurrencyId
WHERE cr.Code = '{}'
AND cp.AccountRole = 'UK.OBIE.Principal'
AND ai.SchemeName LIKE 'UK.%'"""#.format(random.choice(country_list))
#Obtenemos los posibles valores de pais. Se iterará por ellos para cambiar en bucle los registros
country_code_list = ['SPA', 'GBP', 'IND']
random.choice(country_code_list)
from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement
from cassandra import concurrent
from statistics import mean

registers = []
batch = BatchStatement(consistency_level=ConsistencyLevel.ALL)
account_id_list = []

#Cargas Masiva con Many
def aggAmountCurrencyWhenAccountIsEnabledAndSchemeNameIsFromUK():
    
    #Select Currents Account enabled
    SELECT_CURRENTACCOUNT_STMT = """SELECT accountid FROM currentaccountkeyspace.currentaccount WHERE status = 'Enabled' LIMIT 100000 ALLOW FILTERING;"""            
    SELECT_SCHEMENAME_STMT = """SELECT accountid FROM CurrentAccountKeySpace.CurrentAccountbyschemename WHERE schemename LIKE 'UK.%' LIMIT 100000;"""    
    SELECT_POSITION_KEEPING_STMT = """SELECT amount_amount FROM positionkeepingkeyspace.positionkeeping WHERE accountid = '{}';"""    
    
    session = cluster.connect('positionkeepingkeyspace')
    iter = 0;
    i = 1
        
    for i in range(0,repeats): 
                
        time_inicial = time.time()        
        
        enabled_Accounts = []
        uk_scheme_accounts = []
               
        result_ca_enabled = session.execute(SELECT_CURRENTACCOUNT_STMT)
        for accountid in result_ca_enabled:
            enabled_Accounts.append(accountid[0])
        
        #SELECT * FROM positionkeepingkeyspace.positionkeeping where accountid in ('<KEY>', '<KEY>');
        
        result_byscheme = session.execute(SELECT_SCHEMENAME_STMT)
        for accountid in result_byscheme:
            uk_scheme_accounts.append(accountid[0])
        
        #Obtenemos las cuentas que están en ambas listas
        account_to_aggregate = set(enabled_Accounts).intersection(uk_scheme_accounts)
    
        
        #for accountid in account_to_update:
            #print(UPDATE_STMT)    
            #session.execute(UPDATE_STMT,(True, accountid,currency_code,currency_code))
        
        amount_amount_list = []
        
        for accountid in account_to_aggregate:
            result = session.execute (SELECT_POSITION_KEEPING_STMT.format(accountid)) 
            
            if (result[0] == None):
                amount_amount_list.append(0)            
            else:
                amount_amount_list.append(result[0])            
                
        print(amount_amount_list[10:20])
        
        print("max value:", max(amount_amount_list))
        print("min value:", min(amount_amount_list))
        
        
        time_final = time.time()           
        data_time_collection = round(time_final - time_inicial,3)
        used_cpu = psutil.cpu_percent()
        mem_used = psutil.virtual_memory().percent
        registers.append((iter,data_time_collection,used_cpu,mem_used))
        print((iter,data_time_collection,used_cpu,mem_used))
        iter += 1;
        time_inicial = time.time()
        i = i + 1
        
    return registers
registers = aggAmountCurrencyWhenAccountIsEnabledAndSchemeNameIsFromUK()
#Guardamos los resultados Customer Profile
save_results_to_csv(registers,resultados_etl_agregadas)
cluster.shutdown()
print('Conexion cerrada')<jupyter_output>Conexion cerrada
StarcoderdataJupyterStructuredCleanDedup