ApsaraDB for ClickHouse:Análise de compatibilidade e gargalos de desempenho e soluções para migração de ClickHouse autogerenciado para 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é-requisitos

Antes de começar, certifique-se de ter:

• Um ambiente Python 3 no servidor de verificação. Uma instância do Elastic Compute Service (ECS) da Alibaba Cloud com Linux já inclui o Python 3. Caso não use o ECS, instale o Python 3 no site oficial do Python.

• Conectividade de rede do servidor de verificação para ambos os clusters. Para solucionar problemas de conectividade, consulte Como resolvo problemas de conectividade de rede entre o cluster de destino e a fonte de dados?

Verificar compatibilidade de SQL

1. Instale a biblioteca cliente Python para ClickHouse:

   pip3 install clickhouse_driver

2. Crie um arquivo Python com o script a seguir. Substitua os valores de espaço reservado pelos detalhes reais de conexão:

   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. Execute o script e revise a saída. Cada consulta com falha é registrada em vermelho com sua mensagem de erro. Investigue e corrija as falhas com base no erro específico antes de prosseguir com a migração de produção.

Criar um gráfico de chama

1. Conecte-se ao ApsaraDB for ClickHouse usando o clickhouse-client. Para instruções de conexão, consulte Conectar-se a um cluster ClickHouse usando a interface de linha de comando.

2. Exporte o trace_log referente à janela de tempo em que o problema de desempenho ocorreu. Substitua <IP>, <port> e os valores de event_time pelos valores reais:

   Parâmetro	Descrição
   <IP>	Endpoint da Virtual Private Cloud (VPC) do ApsaraDB for ClickHouse
   <port>	Porta TCP do 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 o clickhouse-flamegraph. Para instruções de download e instalação, consulte clickhouse-flamegraph.

4. Gere o gráfico de chama:

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

   Abra o arquivo SVG para identificar as funções que mais consomem CPU. Por exemplo, se ReplacingSortedMerge aparecer com destaque, concentre a investigação em consultas às tabelas ReplacingMergeTree.

   

Analisar query_log

Execute estas consultas simultaneamente em ambos os clusters para realizar uma comparação dos N principais. Procure tabelas nas quais o uso de CPU ou memória seja significativamente maior no ApsaraDB for ClickHouse do que no cluster autogerenciado; essas são as candidatas.

Alto uso de CPU ou memória em uma tabela nem sempre causa degradação geral do cluster. Isso apenas indica maior probabilidade de essa tabela ser a source do problema. Compare os resultados entre ambos os clusters para confirmar.

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

Substitua os valores de event_time pela janela de tempo em que o problema de desempenho ocorreu.

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