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

Prerequisites

Before you begin, ensure that you have:

• A Python 3 environment on the verification server. An Alibaba Cloud Elastic Compute Service (ECS) instance running Linux already includes Python 3. If you're not using ECS, install Python 3 from the Python official website.

• Network connectivity from the verification server to both clusters. For connectivity troubleshooting, see How do I resolve network connectivity issues between the target cluster and the data source?

Verify SQL compatibility

1. Install the Python client library for ClickHouse:

   pip3 install clickhouse_driver

2. Create a Python file with the following script. Replace the placeholder values with your actual connection details:

   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. Run the script and review the output. Each failed query is logged in red with its error message. Investigate and fix failures based on the specific error before proceeding with production migration.

Create a flame graph

1. Connect to ApsaraDB for ClickHouse using clickhouse-client. For connection instructions, see Connect to a ClickHouse cluster using the command-line interface.

2. Export trace_log for the time window when the performance issue occurred. Replace <IP>, <port>, and the event_time values with your actual values:

   Parameter	Description
   <IP>	Virtual Private Cloud (VPC) endpoint of ApsaraDB for ClickHouse
   <port>	TCP port of 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. Install clickhouse-flamegraph. For download and installation instructions, see clickhouse-flamegraph.

4. Generate the flame graph:

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

   Open the SVG file to identify functions consuming the most CPU. For example, if you see ReplacingSortedMerge appearing prominently, focus your investigation on queries against ReplacingMergeTree tables.

   

Analyze query_log

Run these queries simultaneously on both clusters to perform a top N comparison. Look for tables where CPU or memory usage is significantly higher on ApsaraDB for ClickHouse than on the self-managed cluster — those are your candidates.

High CPU or memory usage on a table doesn't always cause overall cluster degradation. It only suggests that table is more likely to be the source of the problem. Compare results between both clusters to confirm.

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

Replace event_time values with the time window when the performance issue occurred.