ApsaraDB for ClickHouse:Compatibility and performance bottleneck analysis and solutions for self-managed ClickHouse migration to ApsaraDB for ClickHouse

SELECT
    name,
    groupArrayDistinct(value) AS value
FROM clusterAllReplicas(`default`, system.settings)
GROUP BY name
ORDER BY name ASC

SET global final = 1;

ALTER TABLE <AIM_TABLE>
    MODIFY SETTING
        min_age_to_force_merge_on_partition_only = 1,
        min_age_to_force_merge_seconds = 60;

Prérequis

Avant de commencer, assurez-vous que :

• Un environnement Python 3 est disponible sur le serveur de vérification. Une instance Alibaba Cloud Elastic Compute Service (ECS) exécutant Linux inclut déjà Python 3. Si vous n'utilisez pas ECS, installez Python 3 depuis le site officiel de Python.

• La connectivité réseau est établie entre le serveur de vérification et les deux clusters. Pour le dépannage de la connectivité, consultez la section Comment résoudre les problèmes de connectivité réseau entre le cluster cible et la source de données ?

Vérifier la compatibilité SQL

1. Installez la bibliothèque cliente Python pour ClickHouse :

   pip3 install clickhouse_driver

2. Créez un fichier Python contenant le script suivant. Remplacez les valeurs d'espace réservé par vos détails de connexion réels :

   from clickhouse_driver import connect
   import datetime
   import logging
   # pip3 install clickhouse_driver is required.
   
   # Connection settings for the self-managed cluster
   host_old = 'HOST_OLD'          # VPC endpoint
   port_old = TCP_PORT_OLD        # TCP port
   user_old = 'USER_OLD'          # Username
   password_old = 'PASSWORD_OLD'  # Password
   
   # Connection settings for ApsaraDB for ClickHouse
   host_new = 'HOST_NEW'          # VPC endpoint
   port_new = TCP_PORT_NEW        # TCP port
   user_new = 'USER_NEW'          # Username
   password_new = 'PASSWORD_NEW'  # Password
   
   logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
   
   def create_connection(host, port, user, password):
       """Establish a connection to ClickHouse."""
       return connect(host=host, port=port, user=user, password=password)
   
   def get_query_hashes(cursor):
       """Query the query_hash list of the last two days."""
       get_queryhash_sql = '''
       select distinct normalized_query_hash from system.query_log
       where type='QueryFinish'
         and `is_initial_query`=1
         and `user` not in ('default', 'aurora')
         and lower(`query`) not like 'select 1%'
         and lower(`query`) not like 'select timezone()%'
         and lower(`query`) not like '%dms-websql%'
         and lower(`query`) like 'select%'
         and `event_time` > now() - INTERVAL 2 DAY;
       '''
       cursor.execute(get_queryhash_sql)
       return cursor.fetchall()
   
   def get_sql_info(cursor, queryhash):
       """Query the SQL information for the specified query_hash in the last 2 days (search range is 3 days)."""
       get_sqlinfo_sql = f'''
       select `event_time`, `query_duration_ms`, `read_rows`, `read_bytes`, `memory_usage`, `databases`, `tables`, `current_database`, `query`
       from system.query_log
       where `event_time` > now() - INTERVAL 3 DAY and `type`='QueryFinish' and `normalized_query_hash`='{queryhash}' limit 1
       '''
       cursor.execute(get_sqlinfo_sql)
       sql_info = cursor.fetchone()
       if sql_info:
           return [info.strftime('%Y-%m-%d %H:%M:%S') if isinstance(info, datetime.datetime) else info for info in sql_info[:-1]], sql_info[-1]
       return None
   
   def execute_sql_on_new_db(cursor, current_database, query_sql, execute_failed_sql):
       """Execute the SQL query on the new cluster and record any failures."""
       try:
           cursor.execute(f"USE {current_database};")
           cursor.execute(query_sql)
       except Exception as error:
           logging.error(f'query_sql execute in new db failed: {query_sql}')
           execute_failed_sql[query_sql] = error
   
   def main():
       conn_old = create_connection(host=host_old, port=port_old, user=user_old, password=password_old)
       conn_new = create_connection(host=host_new, port=port_new, user=user_new, password=password_new)
   
       cursor_old = conn_old.cursor()
       cursor_new = conn_new.cursor()
   
       # Extract query hashes and SQL from the self-managed cluster
       old_query_hashes = get_query_hashes(cursor_old)
       old_db_execute_dir = {}
       for queryhash in old_query_hashes:
           sql_info, query = get_sql_info(cursor_old, queryhash[0])
           if sql_info:
               old_db_execute_dir[query] = sql_info
   
       cursor_old.close()
       conn_old.close()
   
       # Replay queries against ApsaraDB for ClickHouse
       execute_failed_sql = {}
       keys_list = list(old_db_execute_dir.keys())
       for query_sql in old_db_execute_dir:
           position = keys_list.index(query_sql)
           current_database = old_db_execute_dir[query_sql][-1]
           logging.info(f"new db test the {position + 1}th/{len(old_db_execute_dir)}, running sql: {query_sql}\n")
           execute_sql_on_new_db(cursor_new, current_database, query_sql, execute_failed_sql)
   
       # Collect execution results from ApsaraDB for ClickHouse
       new_query_hashes = get_query_hashes(cursor_new)
       new_db_execute_dir = {}
       for queryhash in new_query_hashes:
           sql_info, query = get_sql_info(cursor_new, queryhash[0])
           if sql_info:
               new_db_execute_dir[query] = sql_info
   
       cursor_new.close()
       conn_new.close()
   
       # Log results
       for query_sql in new_db_execute_dir:
           if query_sql in old_db_execute_dir:
               logging.info(f'succeed sql: {query_sql}')
               logging.info(f'old sql info: {old_db_execute_dir[query_sql]}')
               logging.info(f'new sql info: {new_db_execute_dir[query_sql]}\n')
   
       for query_sql in execute_failed_sql:
           logging.error('\033[31m{}\033[0m'.format(f'failed sql: {query_sql}'))
           logging.error('\033[31m{}\033[0m'.format(f'failed error: {execute_failed_sql[query_sql]}\n'))
   
   if __name__ == "__main__":
       main()

3. Exécutez le script et examinez la sortie. Chaque requête ayant échoué est enregistrée en rouge avec son message d'erreur. Analysez et corrigez les échecs en fonction de l'erreur spécifique avant de procéder à la migration de production.

Créer un graphique en flammes

1. Connectez-vous à ApsaraDB for ClickHouse en utilisant clickhouse-client. Pour les instructions de connexion, consultez la section Se connecter à un cluster ClickHouse via l'interface de ligne de commande.

2. Exportez le trace_log pour la fenêtre temporelle durant laquelle le problème de performance s'est produit. Remplacez <IP>, <port> et les valeurs de event_time par vos valeurs réelles :

   Paramètre	Description
   <IP>	Endpoint Virtual Private Cloud (VPC) d'ApsaraDB for ClickHouse
   <port>	Port TCP d'ApsaraDB for ClickHouse

   # trace_type = 'CPU' collects CPU call stacks.
   # trace_type = 'Real' collects wall-clock call stacks.
   /clickhouse/bin/clickhouse-client -h <IP> --port <port> -q \
     "SELECT arrayStringConcat(arrayReverse(arrayMap(x -> concat( addressToLine(x), '#', demangle(addressToSymbol(x)) ), trace)), ';') AS stack, count() AS samples \
     FROM system.trace_log \
     WHERE trace_type = 'CPU' \
       AND event_time >= '2025-01-08 19:31:00' \
       AND event_time < '2025-01-08 19:33:00' \
     GROUP BY trace \
     ORDER BY samples DESC \
     FORMAT TabSeparated \
     SETTINGS allow_introspection_functions=1" > cpu_trace_log.txt

3. Installez clickhouse-flamegraph. Pour les instructions de téléchargement et d'installation, consultez clickhouse-flamegraph.

4. Générez le graphique en flammes :

   cat cpu_trace_log.txt | flamegraph.pl > cpu_trace_log.svg

   Ouvrez le fichier SVG pour identifier les fonctions consommant le plus de CPU. Par exemple, si vous voyez ReplacingSortedMerge apparaître de manière prédominante, concentrez votre enquête sur les requêtes adressées aux tables ReplacingMergeTree.

   

Analyser query_log

Exécutez ces requêtes simultanément sur les deux clusters pour effectuer une comparaison Top N. Recherchez les tables dont l'utilisation du CPU ou de la mémoire est significativement plus élevée sur ApsaraDB for ClickHouse que sur le cluster autogéré ; ce sont vos candidates.

Une utilisation élevée du CPU ou de la mémoire sur une table n'entraîne pas toujours une dégradation globale du cluster. Cela suggère simplement que cette table est plus susceptible d'être à l'origine du problème. Comparez les résultats entre les deux clusters pour confirmer.

SELECT
    tables,
    first_value(query),
    count() AS cnt,
    groupArrayDistinct(normalizedQueryHash(query)) AS normalized_query_hash,
    sum(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'UserTimeMicroseconds')]) AS sum_user_cpu,
    sum(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'SystemTimeMicroseconds')]) AS sum_system_cpu,
    avg(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'UserTimeMicroseconds')]) AS avg_user_cpu,
    avg(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'SystemTimeMicroseconds')]) AS avg_system_cpu,
    sum(memory_usage) AS sum_memory_usage,
    avg(memory_usage) AS avg_memory_usage
FROM clusterAllReplicas(default, system.query_log)
WHERE (event_time > '2025-01-08 19:30:00') AND (event_time < '2025-01-08 20:30:00') AND (query_kind = 'Select')
GROUP BY tables
ORDER BY sum_user_cpu DESC
LIMIT 5

SELECT
    tables,
    first_value(query),
    count() AS cnt,
    groupArrayDistinct(normalizedQueryHash(query)) AS normalized_query_hash,
    sum(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'UserTimeMicroseconds')]) AS sum_user_cpu,
    sum(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'SystemTimeMicroseconds')]) AS sum_system_cpu,
    avg(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'UserTimeMicroseconds')]) AS avg_user_cpu,
    avg(ProfileEvents.Values[indexOf(ProfileEvents.Names, 'SystemTimeMicroseconds')]) AS avg_system_cpu,
    sum(memory_usage) AS sum_memory_usage,
    avg(memory_usage) AS avg_memory_usage
FROM clusterAllReplicas(default, system.query_log)
WHERE (event_time > '2025-01-08 19:30:00') AND (event_time < '2025-01-08 20:30:00') AND (query_kind = 'Select')
GROUP BY tables
ORDER BY sum_memory_usage
LIMIT 5

Remplacez les valeurs de event_time par la fenêtre temporelle durant laquelle le problème de performance s'est produit.

EXPLAIN PIPELINE <SQL with performance issues>

SELECT
    hostname() AS host,
    *
FROM clusterAllReplicas(`default`, system.query_log)
WHERE
    event_time > '2025-01-18 00:00:00'
    AND event_time < '2025-01-18 03:00:00'
    AND initial_query_id = '<INITIAL_QUERY_ID>'
    AND type = 'QueryFinish'
ORDER BY query_start_time_microseconds