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;

Requisitos previos

Antes de empezar, asegúrese de contar con:

• Un entorno de Python 3 en el servidor de verificación. Una instancia de Alibaba Cloud Elastic Compute Service (ECS) con Linux ya incluye Python 3. Si no utiliza ECS, instale Python 3 desde el sitio web oficial de Python.

• Conectividad de red desde el servidor de verificación hacia ambos clústeres. Para solucionar problemas de conectividad, consulte How do I resolve network connectivity issues between the target cluster and the data source?

Verificar la compatibilidad SQL

1. Instale la biblioteca cliente de Python para ClickHouse:

   pip3 install clickhouse_driver

2. Cree un archivo de Python con el siguiente script. Reemplace los valores de marcador de posición con sus detalles de conexión reales:

   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. Ejecute el script y revise la salida. Cada consulta con errores se registra en rojo junto con su mensaje de error. Investigue y corrija los fallos según el error específico antes de proceder con la migración de producción.

Crear un gráfico de llama

1. Conéctese a ApsaraDB for ClickHouse mediante clickhouse-client. Para instrucciones de conexión, consulte Connect to a ClickHouse cluster using the command-line interface.

2. Exporte trace_log correspondiente a la ventana de tiempo en la que ocurrió el problema de rendimiento. Reemplace <IP>, <port> y los valores de event_time con sus valores reales:

   Parámetro	Descripción
   <IP>	Endpoint de Virtual Private Cloud (VPC) de ApsaraDB for ClickHouse
   <port>	Puerto TCP de 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. Instale clickhouse-flamegraph. Para instrucciones de descarga e instalación, consulte clickhouse-flamegraph.

4. Genere el gráfico de llama:

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

   Abra el archivo SVG para identificar las funciones que consumen más CPU. Por ejemplo, si observa que ReplacingSortedMerge aparece de forma destacada, centre su investigación en las consultas dirigidas a tablas ReplacingMergeTree.

   

Analizar query_log

Ejecute estas consultas simultáneamente en ambos clústeres para realizar una comparación de los N principales. Busque tablas donde el uso de CPU o memoria sea significativamente mayor en ApsaraDB for ClickHouse que en el clúster autogestionado; esas son sus candidatas.

Un alto uso de CPU o memoria en una tabla no siempre provoca la degradación general del clúster. Solo indica que es más probable que dicha tabla sea el origen del problema. Compare los resultados entre ambos clústeres para confirmarlo.

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

Reemplace los valores de event_time con la ventana de tiempo en la que ocurrió el problema de rendimiento.

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