{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install pymysql pandas sqlalchemy" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "from sqlalchemy import create_engine\n", "import urllib.parse\n", "\n", "engine = create_engine(f'mysql+pymysql://root:@127.0.0.1:4000/test')\n", "\n", "try:\n", " sql = \"\"\"\n", " SELECT * \n", " FROM prometheus_metrics \n", " \"\"\"\n", " \n", " df = pd.read_sql(sql, engine)\n", " \n", " print(df.info())\n", " print(df.head())\n", " \n", "except Exception as e:\n", " print(f\"Failed to fetch dataset: {e}\")\n", " \n", "finally:\n", " engine.dispose()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "import numpy as np\n", "from datetime import datetime\n", "\n", "def advanced_process_prometheus_data(df, value_tolerance=1e-10, save_problematic=False):\n", " print(f\"Start to process data... time: {datetime.now()}\")\n", " print(f\"Original Data count: {len(df)}\")\n", " print(f\"Unique instance count: {df['prometheus_instance'].nunique()}\")\n", " print(f\"Unique metric count: {df['metric_name'].nunique()}\")\n", " print(f\"Time Range: {df['timestamp'].min()} to {df['timestamp'].max()}\")\n", " \n", " duplicate_analysis = df.groupby(['prometheus_instance', 'timestamp', 'metric_name']).agg({\n", " 'value': ['count', 'nunique', lambda x: list(x.unique())],\n", " 'id': list\n", " }).reset_index()\n", " \n", " duplicate_analysis.columns = ['prometheus_instance', 'timestamp', 'metric_name', \n", " 'count', 'unique_values', 'all_values', 'ids']\n", " \n", " problematic_data = duplicate_analysis[\n", " (duplicate_analysis['count'] > 1) & \n", " (duplicate_analysis['unique_values'] > 1)\n", " ]\n", " \n", " def values_are_different(values_list, tolerance=value_tolerance):\n", " if len(values_list) <= 1:\n", " return False\n", " try:\n", " values_array = np.array(values_list, dtype=float)\n", " return np.ptp(values_array) > tolerance # ptp = max - min\n", " except:\n", " return len(set(values_list)) > 1\n", " \n", " truly_problematic = problematic_data[\n", " problematic_data['all_values'].apply(lambda x: values_are_different(x, value_tolerance))\n", " ]\n", " \n", " print(f\"\\nData Unique Detection:\")\n", " print(f\"- Totally duplicated pair count: {len(duplicate_analysis[duplicate_analysis['count'] > 1])}\")\n", " print(f\"- Same instance and time but different values: {len(truly_problematic)}\")\n", " \n", " problematic_combinations = set()\n", " \n", " if len(truly_problematic) > 0:\n", " print(f\"\\nData issue found:\")\n", " for _, row in truly_problematic.iterrows():\n", " instance = row['prometheus_instance']\n", " timestamp = row['timestamp']\n", " metric = row['metric_name']\n", " values = row['all_values']\n", " \n", " problematic_combinations.add((instance, timestamp))\n", " \n", " print(f\"- Instance: {instance}\")\n", " print(f\" Timestamp: {timestamp}\")\n", " print(f\" Metric: {metric}\")\n", " print(f\" Different values: {values}\")\n", " print(f\" IDs: {row['ids']}\")\n", " print(\"-\" * 30)\n", " \n", " if problematic_combinations:\n", " print(f\"\\nExclude {len(problematic_combinations)} (instance, timestamp) pair...\")\n", " \n", " mask = ~df.set_index(['prometheus_instance', 'timestamp']).index.isin(problematic_combinations)\n", " filtered_df = df[mask].copy()\n", " \n", " if save_problematic:\n", " problematic_df = df[~mask].copy()\n", " timestamp_str = datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n", " filename = f\"problematic_data_{timestamp_str}.csv\"\n", " problematic_df.to_csv(filename, index=False)\n", " print(f\"Excluded data is saved to: {filename}\")\n", " \n", " else:\n", " filtered_df = df.copy()\n", " problematic_df = pd.DataFrame()\n", " \n", " print(f\"Row count after filtering: {len(filtered_df)}\")\n", " \n", " # 3. Handle duplicate rows with identical values\n", " print(\"Removing duplicate rows...\")\n", " deduplicated_df = filtered_df.sort_values('created_at').drop_duplicates(\n", " subset=['prometheus_instance', 'timestamp', 'metric_name'], \n", " keep='last'\n", " )\n", " \n", " print(f\"Row count after deduplication: {len(deduplicated_df)}\")\n", " \n", " # 4. Pivot to wide format\n", " print(\"Converting to wide format...\")\n", " try:\n", " wide_df = deduplicated_df.pivot_table(\n", " index=['prometheus_instance', 'timestamp'],\n", " columns='metric_name',\n", " values='value',\n", " aggfunc='first'\n", " )\n", " \n", " wide_df = wide_df.reset_index()\n", " wide_df.columns.name = None\n", " \n", " # Sort by timestamp\n", " wide_df = wide_df.sort_values(['prometheus_instance', 'timestamp'])\n", " \n", " print(f\"\\nāœ… Conversion completed!\")\n", " print(f\"Final data shape: {wide_df.shape}\")\n", " print(f\"Included metrics: {[col for col in wide_df.columns if col not in ['prometheus_instance', 'timestamp']]}\")\n", " \n", " # Data quality report\n", " print(f\"\\nšŸ“Š Data quality report:\")\n", " print(f\"- Number of instances: {wide_df['prometheus_instance'].nunique()}\")\n", " print(f\"- Number of timestamps: {wide_df['timestamp'].nunique()}\")\n", " print(f\"- Number of metrics: {wide_df.shape[1] - 2}\")\n", " \n", " missing_count = wide_df.isnull().sum().sum()\n", " total_cells = wide_df.shape[0] * (wide_df.shape[1] - 2)\n", " if total_cells > 0:\n", " missing_percentage = (missing_count / total_cells) * 100\n", " print(f\"- Missing values: {missing_count} ({missing_percentage:.2f}%)\")\n", " \n", " return wide_df, problematic_df if save_problematic else None\n", " \n", " except Exception as e:\n", " print(f\"āŒ Pivot to wide format failed: {e}\")\n", " import traceback\n", " traceback.print_exc()\n", " return None, None\n", "\n", "# Example usage\n", "result_df, problematic_data = advanced_process_prometheus_data(\n", " df, \n", " value_tolerance=1e-6, # numeric tolerance\n", " save_problematic=True # save problematic data\n", ")\n", "\n", "if result_df is not None:\n", " # First, report problematic_data (if available)\n", " if problematic_data is None:\n", " print(\"\\nNo problematic_data returned (set save_problematic=True to get it).\")\n", " else:\n", " print(\"\\nProblematic data summary:\")\n", " if problematic_data.empty:\n", " print(\"- No problematic rows were excluded.\")\n", " else:\n", " print(f\"- Excluded rows: {len(problematic_data)}\")\n", " print(f\"- Affected instances: {problematic_data['prometheus_instance'].nunique()}\")\n", " print(f\"- Affected timestamps: {problematic_data['timestamp'].nunique()}\")\n", " print(\"Preview of problematic rows:\")\n", " print(problematic_data.head())\n", "\n", " # Then, show result_df\n", " print(\"\\nPreview of the first few rows of result_df:\")\n", " print(result_df.head())\n", " \n", " # Save result\n", " output_filename = f\"dataset_wide_format.csv\"\n", " result_df.to_csv(output_filename, index=False)\n", " print(f\"\\nResult saved to: {output_filename}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Simple Feature Engineering" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "df = pd.read_csv(\"dataset_wide_format.csv\")" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "============================================================\n", "Processing component: tidb-cpu\n", "============================================================\n", "Available Features: 13/13\n", "Missing Features: set()\n", "=== Feature Engineering Based on TiDB CPU Consumption ===\n", "1. SQL processing complexity characteristics...\n", "2. Memory and CPU interaction characteristics...\n", "3. Workload characteristics...\n", "4. gRPC communication load characteristics...\n", "5. Data quality check...\n", "TiDB CPU Feature Engineering Completed: 13 -> 28 (added: 15)\n", "\n", "=== tidb-cpu Feature Validation ===\n", "Most relevant new features (Top 10):\n", " grpc_complexity_interaction: 0.6813\n", " memory_complexity_interaction: 0.6218\n", " total_expensive_ops: 0.6197\n", " memory_weighted_complexity: 0.6188\n", " weighted_expensive_ops: 0.6172\n", " total_agg_ops: 0.6144\n", " total_client_ops: 0.4569\n", " read_write_ratio: 0.3313\n", " write_intensity: 0.2790\n", " grpc_per_operation: 0.1564\n", "\n", "tidb-cpu Data Preparation Completed:\n", " Feature Matrix: (943, 28)\n", " Target Variable: (943,)\n", " Target Variable Range: [61.95, 467.81]\n", " Target Variable Average: 292.76\n", " New Features Number: 15\n", "\n", "============================================================\n", "Processing component: tikv-cpu\n", "============================================================\n", "Available Features: 15/15\n", "Missing Features: set()\n", "=== Feature Engineering Based on TiKV CPU Consumption ===\n", "1. Storage I/O load characteristics...\n", "2. Memory and storage pressure characteristics...\n", "3. Distributed coordination load characteristics...\n", "4. expensive operations impact on TiKV...\n", "5. Data quality check...\n", "TiKV CPU Feature Engineering Completed: 15 -> 25 (Added: 10)\n", "\n", "=== tikv-cpu Feature Validation ===\n", "Most relevant new features (Top 10):\n", " grpc_io_coordination: 0.6791\n", " memory_io_pressure: 0.6612\n", " tikv_total_io: 0.6089\n", " grpc_per_operation: 0.1126\n", " io_per_operation: 0.1007\n", " io_amplification: 0.1007\n", " memory_per_operation: 0.0854\n", " tikv_weighted_sql_load: 0.0667\n", " tikv_read_write_ratio: 0.0161\n", " sql_io_interaction: 0.0052\n", "\n", "tikv-cpu Data Preparation Completed:\n", " Feature Matrix: (943, 25)\n", " Target Variable: (943,)\n", " Target Variable Range: [0.00, 258.67]\n", " Target Variable Average: 131.63\n", " New Features Number: 10\n", "\n", "============================================================\n", "Processing component: tidb-tikv-total-cpu\n", "============================================================\n", "Available Features: 16/16\n", "Missing Features: set()\n", "=== Feature Engineering Based on Total CPU Consumption of TiDB+TiKV ===\n", "1. SQL processing complexity characteristics (TiDB layer)...\n", "2. Storage I/O load characteristics (TiKV layer)...\n", "3. Distributed coordination load (TiDB-TiKV interaction)...\n", "4. Memory pressure characteristics (dual-layer impact)...\n", "5. Workload characteristics...\n", "6. High-order interaction features...\n", "7. Data quality check...\n", "TiDB+TiKV Total CPU Feature Engineering Completed: 16 -> 40 (Added: 24)\n", "\n", "=== tidb-tikv-total-cpu Feature Validation ===\n", "Most relevant new features (Top 10):\n", " grpc_io_coordination: 0.7080\n", " tikv_total_io: 0.5566\n", " total_client_ops: 0.5391\n", " tikv_memory_io_pressure: 0.4617\n", " read_write_ratio: 0.4240\n", " sql_grpc_interaction: 0.4142\n", " system_load_pressure: 0.4088\n", " write_intensity: 0.3696\n", " tikv_weighted_sql_load: 0.3466\n", " tidb_memory_complexity_load: 0.3414\n", "\n", "tidb-tikv-total-cpu Data Preparation Completed:\n", " Feature Matrix: (943, 40)\n", " Target Variable: (943,)\n", " Target Variable Range: [61.95, 612.23]\n", " Target Variable Average: 424.39\n", " New Features Number: 24\n", "\n", "============================================================\n", "Processing component: gateway-pd-cpu\n", "============================================================\n", "Available Features: 16/16\n", "Missing Features: set()\n", "Skipping gateway-pd-cpu: missing target variable\n", "\n", "šŸŽÆ Final Results:\n", "Target Variable: TiDB CPU + TiKV CPU Total\n", "Feature Number: 40\n", "Sample Number: 943\n", "Target Variable Statistics:\n", " Average: 424.39%\n", " Standard Deviation: 159.31%\n", " Range: [61.95%, 612.23%]\n" ] }, { "data": { "text/plain": [ "Index(['expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up',\n", " 'expensive-index_merge_reader', 'tidb-mem', 'client-read',\n", " 'client-write', 'tikv-read-bytes', 'tikv-write-bytes', 'tikv-mem',\n", " 'tikv-grpc', 'total_expensive_ops', 'tidb_weighted_sql_load',\n", " 'tikv_weighted_sql_load', 'total_join_ops', 'total_agg_ops',\n", " 'tikv_total_io', 'tikv_read_write_ratio', 'io_amplification',\n", " 'grpc_io_coordination', 'grpc_per_operation', 'sql_grpc_interaction',\n", " 'tidb_memory_complexity_load', 'tidb_memory_weighted_load',\n", " 'tikv_memory_io_pressure', 'tikv_memory_sql_pressure',\n", " 'total_memory_pressure', 'memory_balance_ratio', 'total_client_ops',\n", " 'read_write_ratio', 'write_intensity', 'write_amplification',\n", " 'complexity_per_operation', 'system_load_pressure',\n", " 'complex_io_memory_load'],\n", " dtype='object')" ] }, "execution_count": 2, "metadata": {}, "output_type": "execute_result" } ], "source": [ "all_feature_columns = [\n", " 'client-read', 'client-write',\n", " 'conn-dispatch', 'cps', 'expensive-hash_agg', 'expensive-hash_join',\n", " 'expensive-index_look_up', 'expensive-index_look_up_join',\n", " 'expensive-index_merge_reader', 'expensive-merge_join',\n", " 'expensive-nested_loop_apply', 'expensive-sort', 'expensive-stream_agg',\n", " 'pd-cpu', 'tidb-cpu', 'tidb-mem', 'tikv-cpu', 'tikv-grpc', 'tikv-mem',\n", " 'tikv-read-bytes', 'tikv-write-bytes'\n", "]\n", "\n", "import numpy as np\n", "import pandas as pd\n", "from sklearn.preprocessing import StandardScaler\n", "import warnings\n", "warnings.filterwarnings('ignore')\n", "\n", "def tidb_cpu_focused_feature_engineering(X):\n", " \"\"\"\n", " Feature Engineering for Predicting TiDB CPU Usage\n", " \"\"\"\n", " print(\"=== Feature Engineering Based on TiDB CPU Consumption ===\")\n", " X_engineered = X.copy()\n", " new_features = []\n", " \n", " print(\"1. SQL processing complexity characteristics...\")\n", " \n", " # expensive metrics\n", " expensive_ops = [\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " available_expensive = [col for col in expensive_ops if col in X.columns]\n", " \n", " if len(available_expensive) > 0:\n", " # total number of expensive operations\n", " X_engineered['total_expensive_ops'] = X[available_expensive].sum(axis=1)\n", " new_features.append('total_expensive_ops')\n", " \n", " # expensive operations ratio based on client operations\n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_queries = X['client-read'] + X['client-write']\n", " X_engineered['expensive_ops_ratio'] = (X_engineered['total_expensive_ops'] / \n", " (total_queries + 1e-8))\n", " new_features.append('expensive_ops_ratio')\n", " \n", " # weighted expensive operations based on CPU consumption (updated weights)\n", " cpu_weights = {\n", " 'expensive-nested_loop_apply': 4.0, # nested loop apply consumes the most CPU\n", " 'expensive-hash_join': 3.5, # hash join consumes high CPU\n", " 'expensive-merge_join': 3.0, # merge join consumes high CPU \n", " 'expensive-sort': 2.5, # sort consumes high CPU\n", " 'expensive-hash_agg': 2.0, # hash aggregation consumes moderate CPU\n", " 'expensive-stream_agg': 1.8, # stream aggregation consumes moderate CPU\n", " 'expensive-index_look_up_join': 1.5, # index lookup join consumes moderate CPU\n", " 'expensive-index_merge_reader': 1.2, # index merge reader consumes moderate CPU\n", " 'expensive-index_look_up': 1.0 # index lookup consumes moderate CPU\n", " }\n", " \n", " weighted_expensive = 0\n", " for op in available_expensive:\n", " if op in cpu_weights:\n", " weighted_expensive += X[op] * cpu_weights[op]\n", " \n", " X_engineered['weighted_expensive_ops'] = weighted_expensive\n", " new_features.append('weighted_expensive_ops')\n", " \n", " # JOIN operation characteristics (merging different types of JOINs)\n", " join_ops = ['expensive-merge_join', 'expensive-hash_join', \n", " 'expensive-nested_loop_apply', 'expensive-index_look_up_join']\n", " available_joins = [col for col in join_ops if col in X.columns]\n", " \n", " if len(available_joins) > 0:\n", " X_engineered['total_join_ops'] = X[available_joins].sum(axis=1)\n", " X_engineered['join_ops_ratio'] = (X_engineered['total_join_ops'] / \n", " (X_engineered['total_expensive_ops'] + 1e-8))\n", " new_features.extend(['total_join_ops', 'join_ops_ratio'])\n", " \n", " # aggregation operation characteristics\n", " agg_ops = ['expensive-hash_agg', 'expensive-stream_agg']\n", " available_aggs = [col for col in agg_ops if col in X.columns]\n", " \n", " if len(available_aggs) > 0:\n", " X_engineered['total_agg_ops'] = X[available_aggs].sum(axis=1)\n", " X_engineered['agg_ops_ratio'] = (X_engineered['total_agg_ops'] / \n", " (X_engineered['total_expensive_ops'] + 1e-8))\n", " new_features.extend(['total_agg_ops', 'agg_ops_ratio'])\n", " \n", " # ==================== 2. Memory and CPU interaction characteristics ====================\n", " print(\"2. Memory and CPU interaction characteristics...\")\n", " \n", " if 'tidb-mem' in X.columns:\n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['memory_complexity_interaction'] = (X['tidb-mem'] * \n", " X_engineered['total_expensive_ops'])\n", " new_features.append('memory_complexity_interaction')\n", " \n", " if 'weighted_expensive_ops' in X_engineered.columns:\n", " X_engineered['memory_weighted_complexity'] = (X['tidb-mem'] * \n", " X_engineered['weighted_expensive_ops'])\n", " new_features.append('memory_weighted_complexity')\n", " \n", " # ==================== 3. Workload characteristics ====================\n", " print(\"3. Workload characteristics...\")\n", " \n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_ops = X['client-read'] + X['client-write']\n", " X_engineered['read_write_ratio'] = (X['client-read'] / (X['client-write'] + 1e-8))\n", " X_engineered['total_client_ops'] = total_ops\n", " X_engineered['write_intensity'] = X['client-write'] / (total_ops + 1e-8)\n", " new_features.extend(['read_write_ratio', 'total_client_ops', 'write_intensity'])\n", " \n", " # average expensive complexity per client operation\n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['complexity_per_operation'] = (X_engineered['total_expensive_ops'] / \n", " (total_ops + 1e-8))\n", " new_features.append('complexity_per_operation')\n", " \n", " # ==================== 4. gRPC communication load characteristics ====================\n", " print(\"4. gRPC communication load characteristics...\")\n", " \n", " if 'tikv-grpc' in X.columns:\n", " if 'total_client_ops' in X_engineered.columns:\n", " X_engineered['grpc_per_operation'] = (X['tikv-grpc'] / \n", " (X_engineered['total_client_ops'] + 1e-8))\n", " new_features.append('grpc_per_operation')\n", " \n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['grpc_complexity_interaction'] = (X['tikv-grpc'] * \n", " X_engineered['total_expensive_ops'])\n", " new_features.append('grpc_complexity_interaction')\n", " \n", " # ==================== 5. Data quality check ====================\n", " print(\"5. Data quality check...\")\n", " X_engineered = X_engineered.replace([np.inf, -np.inf], np.nan)\n", " for feature in new_features:\n", " if feature in X_engineered.columns and X_engineered[feature].isnull().sum() > 0:\n", " X_engineered[feature] = X_engineered[feature].fillna(X_engineered[feature].median())\n", " \n", " print(f\"TiDB CPU Feature Engineering Completed: {X.shape[1]} -> {X_engineered.shape[1]} (added: {len(new_features)})\")\n", " return X_engineered, new_features\n", "\n", "def tikv_cpu_focused_feature_engineering(X):\n", " \"\"\"\n", " Feature Engineering for Predicting TiKV Server CPU Usage\n", " \"\"\"\n", " print(\"=== Feature Engineering Based on TiKV CPU Consumption ===\")\n", " X_engineered = X.copy()\n", " new_features = []\n", " \n", " # ==================== 1. Storage I/O load characteristics ====================\n", " print(\"1. Storage I/O load characteristics...\")\n", " \n", " io_features = ['tikv-read-bytes', 'tikv-write-bytes']\n", " available_io = [col for col in io_features if col in X.columns]\n", " \n", " if len(available_io) >= 2:\n", " X_engineered['tikv_total_io'] = X[available_io].sum(axis=1)\n", " X_engineered['tikv_read_write_ratio'] = (X['tikv-read-bytes'] / \n", " (X['tikv-write-bytes'] + 1e-8))\n", " new_features.extend(['tikv_total_io', 'tikv_read_write_ratio'])\n", " \n", " # I/O density (relative to client operations)\n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_client_ops = X['client-read'] + X['client-write']\n", " X_engineered['io_per_operation'] = (X_engineered['tikv_total_io'] / \n", " (total_client_ops + 1e-8))\n", " new_features.append('io_per_operation')\n", " \n", " # I/O amplification factor\n", " X_engineered['io_amplification'] = (X_engineered['tikv_total_io'] / \n", " (total_client_ops + 1e-8))\n", " new_features.append('io_amplification')\n", " \n", " # ==================== 2. Memory and storage pressure characteristics ====================\n", " print(\"2. Memory and storage pressure characteristics...\")\n", " \n", " if 'tikv-mem' in X.columns:\n", " if 'tikv_total_io' in X_engineered.columns:\n", " X_engineered['memory_io_pressure'] = (X['tikv-mem'] * \n", " X_engineered['tikv_total_io'])\n", " new_features.append('memory_io_pressure')\n", " \n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_ops = X['client-read'] + X['client-write']\n", " X_engineered['memory_per_operation'] = (X['tikv-mem'] / (total_ops + 1e-8))\n", " new_features.append('memory_per_operation')\n", " \n", " # ==================== 3. Distributed coordination load characteristics ====================\n", " print(\"3. Distributed coordination load characteristics...\")\n", " \n", " if 'tikv-grpc' in X.columns:\n", " if 'tikv_total_io' in X_engineered.columns:\n", " X_engineered['grpc_io_coordination'] = (X['tikv-grpc'] * \n", " X_engineered['tikv_total_io'])\n", " new_features.append('grpc_io_coordination')\n", " \n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_ops = X['client-read'] + X['client-write']\n", " X_engineered['grpc_per_operation'] = (X['tikv-grpc'] / (total_ops + 1e-8))\n", " new_features.append('grpc_per_operation')\n", " \n", " # ==================== 4. expensive operations impact on TiKV ====================\n", " print(\"4. expensive operations impact on TiKV...\")\n", " \n", " # check if there is data for expensive operations\n", " expensive_ops = [\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " available_expensive = [col for col in expensive_ops if col in X.columns]\n", " \n", " if len(available_expensive) > 0:\n", " # operations that have a large impact on the TiKV storage layer\n", " tikv_impact_weights = {\n", " 'expensive-index_look_up': 3.0, # index lookup has a large impact on storage\n", " 'expensive-index_look_up_join': 2.5, # index lookup join\n", " 'expensive-index_merge_reader': 2.0, # index merge read\n", " 'expensive-merge_join': 1.8, # Merge Join needs to sort data\n", " 'expensive-hash_join': 1.5, # Hash Join has a small impact\n", " 'expensive-sort': 1.2, # Sorting may require external storage\n", " 'expensive-hash_agg': 1.0, # Hash aggregation operation\n", " 'expensive-stream_agg': 0.8, # Stream aggregation has a small impact\n", " 'expensive-nested_loop_apply': 2.5 # Nested loop may read a lot\n", " }\n", " \n", " tikv_weighted_load = 0\n", " for op in available_expensive:\n", " if op in tikv_impact_weights:\n", " tikv_weighted_load += X[op] * tikv_impact_weights[op]\n", " \n", " X_engineered['tikv_weighted_sql_load'] = tikv_weighted_load\n", " new_features.append('tikv_weighted_sql_load')\n", " \n", " # SQL complexity and I/O interaction\n", " if 'tikv_total_io' in X_engineered.columns:\n", " X_engineered['sql_io_interaction'] = (tikv_weighted_load * \n", " X_engineered['tikv_total_io'])\n", " new_features.append('sql_io_interaction')\n", " \n", " # ==================== 5. Data quality check ====================\n", " print(\"5. Data quality check...\")\n", " X_engineered = X_engineered.replace([np.inf, -np.inf], np.nan)\n", " for feature in new_features:\n", " if feature in X_engineered.columns and X_engineered[feature].isnull().sum() > 0:\n", " X_engineered[feature] = X_engineered[feature].fillna(X_engineered[feature].median())\n", " \n", " print(f\"TiKV CPU Feature Engineering Completed: {X.shape[1]} -> {X_engineered.shape[1]} (Added: {len(new_features)})\")\n", " return X_engineered, new_features\n", "\n", "def tidb_tikv_total_cpu_feature_engineering(X):\n", " \"\"\"\n", " Feature Engineering for Predicting Total CPU Usage of TiDB+TiKV\n", " \"\"\"\n", " print(\"=== Feature Engineering Based on Total CPU Consumption of TiDB+TiKV ===\")\n", " X_engineered = X.copy()\n", " new_features = []\n", " \n", " # ==================== 1. SQL processing complexity characteristics (TiDB layer) ====================\n", " print(\"1. SQL processing complexity characteristics (TiDB layer)...\")\n", " \n", " expensive_ops = [\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " available_expensive = [col for col in expensive_ops if col in X.columns]\n", " \n", " if len(available_expensive) > 0:\n", " # TiDB SQL processing load\n", " X_engineered['total_expensive_ops'] = X[available_expensive].sum(axis=1)\n", " new_features.append('total_expensive_ops')\n", " \n", " # Weighted SQL complexity (TiDB CPU weights)\n", " tidb_cpu_weights = {\n", " 'expensive-nested_loop_apply': 4.0, # Nested loop consumes the most TiDB CPU\n", " 'expensive-hash_join': 3.5, # Hash Join consumes high CPU\n", " 'expensive-merge_join': 3.0, # Merge Join consumes high CPU \n", " 'expensive-sort': 2.5, # Sorting consumes high CPU\n", " 'expensive-hash_agg': 2.0, # Hash aggregation consumes moderate CPU\n", " 'expensive-stream_agg': 1.8, # Stream aggregation consumes moderate CPU\n", " 'expensive-index_look_up_join': 1.5, # Index lookup join consumes moderate CPU\n", " 'expensive-index_merge_reader': 1.2, # Index merge read consumes moderate CPU\n", " 'expensive-index_look_up': 1.0 # Simple index lookup\n", " }\n", " \n", " tidb_weighted_ops = 0\n", " for op in available_expensive:\n", " if op in tidb_cpu_weights:\n", " tidb_weighted_ops += X[op] * tidb_cpu_weights[op]\n", " \n", " X_engineered['tidb_weighted_sql_load'] = tidb_weighted_ops\n", " new_features.append('tidb_weighted_sql_load')\n", " \n", " # TiKV storage layer impact on SQL complexity (different weights)\n", " tikv_storage_weights = {\n", " 'expensive-index_look_up': 3.0, # Index lookup consumes the most storage\n", " 'expensive-index_look_up_join': 2.5, # Index lookup JOIN\n", " 'expensive-index_merge_reader': 2.2, # Index merge read\n", " 'expensive-merge_join': 2.0, # Merge Join needs to sort data\n", " 'expensive-nested_loop_apply': 2.5, # Nested loop reads a lot\n", " 'expensive-hash_join': 1.5, # Hash Join has a small impact\n", " 'expensive-sort': 1.2, # Sorting may require external storage\n", " 'expensive-hash_agg': 1.0, # Hash aggregation needs to read data\n", " 'expensive-stream_agg': 0.8 # Stream aggregation has a small impact\n", " }\n", " \n", " tikv_weighted_ops = 0\n", " for op in available_expensive:\n", " if op in tikv_storage_weights:\n", " tikv_weighted_ops += X[op] * tikv_storage_weights[op]\n", " \n", " X_engineered['tikv_weighted_sql_load'] = tikv_weighted_ops\n", " new_features.append('tikv_weighted_sql_load')\n", " \n", " # JOIN and aggregation operation classification\n", " join_ops = ['expensive-merge_join', 'expensive-hash_join', \n", " 'expensive-nested_loop_apply', 'expensive-index_look_up_join']\n", " available_joins = [col for col in join_ops if col in X.columns]\n", " \n", " if len(available_joins) > 0:\n", " X_engineered['total_join_ops'] = X[available_joins].sum(axis=1)\n", " new_features.append('total_join_ops')\n", " \n", " agg_ops = ['expensive-hash_agg', 'expensive-stream_agg']\n", " available_aggs = [col for col in agg_ops if col in X.columns]\n", " \n", " if len(available_aggs) > 0:\n", " X_engineered['total_agg_ops'] = X[available_aggs].sum(axis=1)\n", " new_features.append('total_agg_ops')\n", " \n", " # ==================== 2. Storage I/O load characteristics (TiKV layer) ====================\n", " print(\"2. Storage I/O load characteristics (TiKV layer)...\")\n", " \n", " io_features = ['tikv-read-bytes', 'tikv-write-bytes']\n", " available_io = [col for col in io_features if col in X.columns]\n", " \n", " if len(available_io) >= 2:\n", " # Total I/O load\n", " X_engineered['tikv_total_io'] = X[available_io].sum(axis=1)\n", " X_engineered['tikv_read_write_ratio'] = (X['tikv-read-bytes'] / \n", " (X['tikv-write-bytes'] + 1e-8))\n", " new_features.extend(['tikv_total_io', 'tikv_read_write_ratio'])\n", " \n", " # I/O efficiency metrics\n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_client_ops = X['client-read'] + X['client-write']\n", " X_engineered['io_amplification'] = (X_engineered['tikv_total_io'] / \n", " (total_client_ops + 1e-8))\n", " new_features.append('io_amplification')\n", " \n", " # ==================== 3. Distributed coordination load (TiDB-TiKV interaction) ====================\n", " print(\"3. Distributed coordination load (TiDB-TiKV interaction)...\")\n", " \n", " if 'tikv-grpc' in X.columns:\n", " # gRPC communication load (affects both sides)\n", " if 'tikv_total_io' in X_engineered.columns:\n", " X_engineered['grpc_io_coordination'] = (X['tikv-grpc'] * \n", " X_engineered['tikv_total_io'])\n", " new_features.append('grpc_io_coordination')\n", " \n", " # Communication efficiency\n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_ops = X['client-read'] + X['client-write']\n", " X_engineered['grpc_per_operation'] = (X['tikv-grpc'] / (total_ops + 1e-8))\n", " new_features.append('grpc_per_operation')\n", " \n", " # SQL complexity Ɨ gRPC coordination (cross-layer interaction)\n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['sql_grpc_interaction'] = (X_engineered['total_expensive_ops'] * \n", " X['tikv-grpc'])\n", " new_features.append('sql_grpc_interaction')\n", " \n", " # ==================== 4. Memory pressure characteristics (dual-layer impact) ====================\n", " print(\"4. Memory pressure characteristics (dual-layer impact)...\")\n", " \n", " # TiDB memory pressure\n", " if 'tidb-mem' in X.columns:\n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['tidb_memory_complexity_load'] = (X['tidb-mem'] * \n", " X_engineered['total_expensive_ops'])\n", " new_features.append('tidb_memory_complexity_load')\n", " \n", " if 'tidb_weighted_sql_load' in X_engineered.columns:\n", " X_engineered['tidb_memory_weighted_load'] = (X['tidb-mem'] * \n", " X_engineered['tidb_weighted_sql_load'])\n", " new_features.append('tidb_memory_weighted_load')\n", " \n", " # TiKV memory pressure\n", " if 'tikv-mem' in X.columns:\n", " if 'tikv_total_io' in X_engineered.columns:\n", " X_engineered['tikv_memory_io_pressure'] = (X['tikv-mem'] * \n", " X_engineered['tikv_total_io'])\n", " new_features.append('tikv_memory_io_pressure')\n", " \n", " if 'tikv_weighted_sql_load' in X_engineered.columns:\n", " X_engineered['tikv_memory_sql_pressure'] = (X['tikv-mem'] * \n", " X_engineered['tikv_weighted_sql_load'])\n", " new_features.append('tikv_memory_sql_pressure')\n", " \n", " # Total memory pressure\n", " if 'tidb-mem' in X.columns and 'tikv-mem' in X.columns:\n", " X_engineered['total_memory_pressure'] = X['tidb-mem'] + X['tikv-mem']\n", " X_engineered['memory_balance_ratio'] = (X['tidb-mem'] / \n", " (X['tikv-mem'] + 1e-8))\n", " new_features.extend(['total_memory_pressure', 'memory_balance_ratio'])\n", " \n", " # ==================== 5. Workload characteristics ====================\n", " print(\"5. Workload characteristics...\")\n", " \n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_ops = X['client-read'] + X['client-write']\n", " X_engineered['total_client_ops'] = total_ops\n", " X_engineered['read_write_ratio'] = (X['client-read'] / (X['client-write'] + 1e-8))\n", " X_engineered['write_intensity'] = X['client-write'] / (total_ops + 1e-8)\n", " new_features.extend(['total_client_ops', 'read_write_ratio', 'write_intensity'])\n", " \n", " # Write operations affect both layers\n", " if 'tikv-write-bytes' in X.columns:\n", " X_engineered['write_amplification'] = (X['tikv-write-bytes'] / \n", " (X['client-write'] + 1e-8))\n", " new_features.append('write_amplification')\n", " \n", " # The complexity of each operation\n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['complexity_per_operation'] = (X_engineered['total_expensive_ops'] / \n", " (total_ops + 1e-8))\n", " new_features.append('complexity_per_operation')\n", " \n", " # ==================== 6. High-order interaction features ====================\n", " print(\"6. High-order interaction features...\")\n", " \n", " # System overall load (affects both layers)\n", " if ('total_client_ops' in X_engineered.columns and \n", " 'tikv_total_io' in X_engineered.columns and\n", " 'total_expensive_ops' in X_engineered.columns):\n", " \n", " X_engineered['system_load_pressure'] = (\n", " X_engineered['total_client_ops'] * \n", " X_engineered['tikv_total_io'] * \n", " X_engineered['total_expensive_ops']\n", " ) ** (1/3) # Geometric mean\n", " new_features.append('system_load_pressure')\n", " \n", " # SQL complexity Ɨ I/O load Ɨ memory pressure (three-dimensional interaction)\n", " if ('total_expensive_ops' in X_engineered.columns and \n", " 'tikv_total_io' in X_engineered.columns and \n", " 'total_memory_pressure' in X_engineered.columns):\n", " \n", " X_engineered['complex_io_memory_load'] = (\n", " X_engineered['total_expensive_ops'] * \n", " X_engineered['tikv_total_io'] * \n", " X_engineered['total_memory_pressure']\n", " ) ** (1/3)\n", " new_features.append('complex_io_memory_load')\n", " \n", " # ==================== 7. Data quality check ====================\n", " print(\"7. Data quality check...\")\n", " X_engineered = X_engineered.replace([np.inf, -np.inf], np.nan)\n", " for feature in new_features:\n", " if feature in X_engineered.columns and X_engineered[feature].isnull().sum() > 0:\n", " X_engineered[feature] = X_engineered[feature].fillna(X_engineered[feature].median())\n", " \n", " print(f\"TiDB+TiKV Total CPU Feature Engineering Completed: {X.shape[1]} -> {X_engineered.shape[1]} (Added: {len(new_features)})\")\n", " return X_engineered, new_features\n", "\n", "def gateway_pd_cpu_focused_feature_engineering(X):\n", " \"\"\"\n", " Feature Engineering for Predicting CPU Usage of Gateway+PD (Updated Version)\n", " \"\"\"\n", " print(\"=== Feature Engineering Based on Gateway+PD CPU Consumption (New Data Version) ===\")\n", " X_engineered = X.copy()\n", " new_features = []\n", " \n", " # ==================== 1. Request routing load characteristics ====================\n", " print(\"1. Request routing load characteristics...\")\n", " \n", " if 'client-read' in X.columns and 'client-write' in X.columns:\n", " total_requests = X['client-read'] + X['client-write']\n", " X_engineered['total_routing_load'] = total_requests\n", " X_engineered['read_write_ratio'] = (X['client-read'] / (X['client-write'] + 1e-8))\n", " X_engineered['write_intensity'] = X['client-write'] / (total_requests + 1e-8)\n", " new_features.extend(['total_routing_load', 'read_write_ratio', 'write_intensity'])\n", " \n", " # ==================== 2. System load balance characteristics ====================\n", " print(\"2. System load balance characteristics...\")\n", " \n", " system_load_components = []\n", " if 'total_routing_load' in X_engineered.columns:\n", " system_load_components.append('total_routing_load')\n", " \n", " # Cluster resource features\n", " resource_features = ['tikv-mem', 'tidb-mem']\n", " available_resources = [col for col in resource_features if col in X.columns]\n", " if len(available_resources) >= 2:\n", " X_engineered['cluster_memory_total'] = X[available_resources].sum(axis=1)\n", " X_engineered['memory_balance_ratio'] = (X['tidb-mem'] / (X['tikv-mem'] + 1e-8))\n", " system_load_components.append('cluster_memory_total')\n", " new_features.extend(['cluster_memory_total', 'memory_balance_ratio'])\n", " \n", " # Cluster I/O features\n", " io_features = ['tikv-read-bytes', 'tikv-write-bytes']\n", " available_io = [col for col in io_features if col in X.columns]\n", " if len(available_io) >= 2:\n", " X_engineered['total_cluster_io'] = X[available_io].sum(axis=1)\n", " system_load_components.append('total_cluster_io')\n", " new_features.append('total_cluster_io')\n", " \n", " # Overall system pressure\n", " if len(system_load_components) >= 2:\n", " X_engineered['overall_system_pressure'] = np.power(\n", " X_engineered[system_load_components].prod(axis=1), \n", " 1/len(system_load_components)\n", " )\n", " new_features.append('overall_system_pressure')\n", " \n", " # ==================== 3. Distributed coordination load ====================\n", " print(\"3. Distributed coordination load...\")\n", " \n", " if 'tikv-grpc' in X.columns:\n", " if 'total_routing_load' in X_engineered.columns:\n", " X_engineered['grpc_routing_coordination'] = (X['tikv-grpc'] * \n", " X_engineered['total_routing_load'])\n", " new_features.append('grpc_routing_coordination')\n", " \n", " if 'total_cluster_io' in X_engineered.columns:\n", " X_engineered['grpc_io_coordination'] = (X['tikv-grpc'] * \n", " X_engineered['total_cluster_io'])\n", " new_features.append('grpc_io_coordination')\n", " \n", " # ==================== 4. The scheduling impact of expensive operations ====================\n", " print(\"4. The scheduling impact of expensive operations...\")\n", " \n", " expensive_ops = [\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " available_expensive = [col for col in expensive_ops if col in X.columns]\n", " \n", " if len(available_expensive) > 0:\n", " X_engineered['total_expensive_ops'] = X[available_expensive].sum(axis=1)\n", " new_features.append('total_expensive_ops')\n", " \n", " # The scheduling impact of expensive operations\n", " if 'total_routing_load' in X_engineered.columns:\n", " X_engineered['expensive_routing_pressure'] = (X_engineered['total_expensive_ops'] * \n", " X_engineered['total_routing_load'])\n", " new_features.append('expensive_routing_pressure')\n", " \n", " # Resource coordination for complex queries\n", " if 'cluster_memory_total' in X_engineered.columns:\n", " X_engineered['expensive_resource_coordination'] = (X_engineered['total_expensive_ops'] * \n", " X_engineered['cluster_memory_total'])\n", " new_features.append('expensive_resource_coordination')\n", " \n", " # ==================== 5. Transaction coordination load ====================\n", " print(\"5. Transaction coordination load...\")\n", " \n", " if 'client-write' in X.columns:\n", " # Write operations require more coordination\n", " if 'tikv-write-bytes' in X.columns:\n", " X_engineered['write_coordination_load'] = (X['client-write'] * \n", " X['tikv-write-bytes'])\n", " new_features.append('write_coordination_load')\n", " \n", " if 'total_expensive_ops' in X_engineered.columns:\n", " X_engineered['write_expensive_coordination'] = (X['client-write'] * \n", " X_engineered['total_expensive_ops'])\n", " new_features.append('write_expensive_coordination')\n", " \n", " # ==================== 6. Data quality check ====================\n", " print(\"6. Data quality check...\")\n", " X_engineered = X_engineered.replace([np.inf, -np.inf], np.nan)\n", " for feature in new_features:\n", " if feature in X_engineered.columns and X_engineered[feature].isnull().sum() > 0:\n", " X_engineered[feature] = X_engineered[feature].fillna(X_engineered[feature].median())\n", " \n", " print(f\"Gateway+PD CPU Feature Engineering Completed: {X.shape[1]} -> {X_engineered.shape[1]} (Added: {len(new_features)})\")\n", " return X_engineered, new_features\n", "\n", "def validate_features(X_engineered, new_features, y, component_name):\n", " \"\"\"\n", " Validate the effectiveness of feature engineering\n", " \"\"\"\n", " print(f\"\\n=== {component_name} Feature Validation ===\")\n", " \n", " # Calculate the correlation between new features and the target variable\n", " new_feature_corrs = []\n", " for feature in new_features:\n", " if feature in X_engineered.columns:\n", " corr = abs(X_engineered[feature].corr(y))\n", " if not np.isnan(corr):\n", " new_feature_corrs.append((feature, corr))\n", " \n", " new_feature_corrs.sort(key=lambda x: x[1], reverse=True)\n", " print(f\"Most relevant new features (Top 10):\")\n", " for feature, corr in new_feature_corrs[:10]:\n", " print(f\" {feature}: {corr:.4f}\")\n", " \n", " return new_feature_corrs\n", "\n", "# Component feature configuration (remove connection-related metrics, use new expensive operations)\n", "tidb_component_features = {\n", " \"tidb-cpu\": {\n", " \"features\": [\n", " # expensive operation metrics\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader',\n", " # Other features\n", " 'tidb-mem', 'client-read', 'client-write', 'tikv-grpc'\n", " ]\n", " },\n", " \"tikv-cpu\": {\n", " \"features\": [\n", " 'tikv-read-bytes', 'tikv-write-bytes', 'tikv-mem', 'tikv-grpc',\n", " 'client-read', 'client-write',\n", " # expensive operation metrics\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " },\n", " \"tidb-tikv-total-cpu\": { # TiDB+TiKV Total CPU\n", " \"features\": [\n", " # TiDB layer features\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader',\n", " 'tidb-mem', 'client-read', 'client-write',\n", " # TiKV layer features \n", " 'tikv-read-bytes', 'tikv-write-bytes', 'tikv-mem', 'tikv-grpc'\n", " ]\n", " },\n", " \"gateway-pd-cpu\": {\n", " \"features\": [\n", " 'client-read', 'client-write', 'tikv-grpc', 'tikv-read-bytes', \n", " 'tikv-write-bytes', 'tikv-mem', 'tidb-mem',\n", " # expensive operation metrics\n", " 'expensive-merge_join', 'expensive-hash_join', 'expensive-hash_agg',\n", " 'expensive-stream_agg', 'expensive-sort', 'expensive-nested_loop_apply',\n", " 'expensive-index_look_up_join', 'expensive-index_look_up', 'expensive-index_merge_reader'\n", " ]\n", " }\n", "}\n", "\n", "def run_component_feature_engineering(df, component_features):\n", " results = {}\n", " \n", " for target, config in component_features.items():\n", " print(f\"\\n{'='*60}\")\n", " print(f\"Processing component: {target}\")\n", " print(f\"{'='*60}\")\n", " \n", " available_features = [f for f in config['features'] if f in df.columns]\n", " print(f\"Available Features: {len(available_features)}/{len(config['features'])}\")\n", " print(f\"Missing Features: {set(config['features']) - set(available_features)}\")\n", " \n", " X = df[available_features]\n", " \n", " if target == 'tidb-cpu':\n", " y = df['tidb-cpu']\n", " X_eng, new_features = tidb_cpu_focused_feature_engineering(X)\n", " elif target == 'tikv-cpu':\n", " y = df['tikv-cpu'] \n", " X_eng, new_features = tikv_cpu_focused_feature_engineering(X)\n", " elif target == 'tidb-tikv-total-cpu': # TiDB+TiKV Total CPU\n", " y = df['tidb-cpu'] + df['tikv-cpu'] # Target variable: TiDB + TiKV Total CPU\n", " X_eng, new_features = tidb_tikv_total_cpu_feature_engineering(X)\n", " elif target == 'gateway-pd-cpu':\n", " if 'gateway-cpu' in df.columns and 'pd-cpu' in df.columns:\n", " y = df['gateway-cpu'] + df['pd-cpu']\n", " else:\n", " print(f\"Skipping {target}: missing target variable\")\n", " continue\n", " X_eng, new_features = gateway_pd_cpu_focused_feature_engineering(X)\n", " \n", " # Validate features\n", " correlations = validate_features(X_eng, new_features, y, target)\n", " \n", " results[target] = {\n", " 'X': X_eng,\n", " 'y': y,\n", " 'new_features': new_features,\n", " 'correlations': correlations\n", " }\n", " \n", " print(f\"\\n{target} Data Preparation Completed:\")\n", " print(f\" Feature Matrix: {X_eng.shape}\")\n", " print(f\" Target Variable: {y.shape}\")\n", " print(f\" Target Variable Range: [{y.min():.2f}, {y.max():.2f}]\")\n", " print(f\" Target Variable Average: {y.mean():.2f}\")\n", " print(f\" New Features Number: {len(new_features)}\")\n", " \n", " return results\n", "\n", "\n", "results = run_component_feature_engineering(df, tidb_component_features)\n", "\n", "# # Fetch the results of TiDB+TiKV Total CPU\n", "target_label = \"tidb-tikv-total-cpu\"\n", "y = results[target_label]['y']\n", "X = results[target_label]['X'] \n", "selected_feature_columns = X.columns\n", "selected_expert_features = X.columns\n", "\n", "print(f\"\\nšŸŽÆ Final Results:\")\n", "print(f\"Target Variable: TiDB CPU + TiKV CPU Total\")\n", "print(f\"Feature Number: {len(selected_expert_features)}\")\n", "print(f\"Sample Number: {len(y)}\")\n", "print(f\"Target Variable Statistics:\")\n", "print(f\" Average: {y.mean():.2f}%\")\n", "print(f\" Standard Deviation: {y.std():.2f}%\") \n", "print(f\" Range: [{y.min():.2f}%, {y.max():.2f}%]\")\n", "\n", "selected_expert_features\n" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "šŸ’¾ Saving datasets to 'dataset/' directory...\n", "āœ… Saved tidb-cpu: (943, 30)\n", "āœ… Saved tikv-cpu: (943, 27)\n", "āœ… Saved tidb-tikv-total-cpu: (943, 42)\n", "\n", "āœ… All datasets saved:\n", "šŸ“ Directory: dataset/\n", "šŸ“„ Files: 3 dataset files + 1 metadata file\n", "šŸ“‹ Metadata: dataset/metadata.json\n", "\n", "šŸ“‚ Dataset directory contents:\n", " tikv-cpu.pkl: (943, 27)\n", " metadata.json\n", " tidb-tikv-total-cpu.pkl: (943, 42)\n", " tidb-cpu.pkl: (943, 30)\n" ] } ], "source": [ "import os\n", "import json\n", "from datetime import datetime\n", "\n", "# Create dataset directory\n", "dataset_dir = \"dataset\"\n", "os.makedirs(dataset_dir, exist_ok=True)\n", "\n", "print(f\"\\nšŸ’¾ Saving datasets to '{dataset_dir}/' directory...\")\n", "\n", "# Save each component dataset\n", "metadata = {\n", " 'created_at': datetime.now().isoformat(),\n", " 'total_components': len(results),\n", " 'components': {}\n", "}\n", "\n", "saved_files = []\n", "\n", "for target_name, data in results.items():\n", " # Combine features and target\n", " dataset = data['X'].copy()\n", " dataset['target'] = data['y']\n", " dataset['target_type'] = target_name\n", " \n", " # Save dataset file\n", " dataset_file = os.path.join(dataset_dir, f\"{target_name}.pkl\")\n", " dataset.to_pickle(dataset_file)\n", " saved_files.append(dataset_file)\n", " \n", " # Collect metadata\n", " metadata['components'][target_name] = {\n", " 'file': f\"{target_name}.pkl\",\n", " 'sample_count': len(data['y']),\n", " 'feature_count': len(data['X'].columns),\n", " 'new_features_count': len(data['new_features']),\n", " 'new_features': data['new_features'],\n", " 'target_stats': {\n", " 'mean': float(data['y'].mean()),\n", " 'std': float(data['y'].std()),\n", " 'min': float(data['y'].min()),\n", " 'max': float(data['y'].max())\n", " },\n", " 'shape': list(dataset.shape)\n", " }\n", " \n", " print(f\"āœ… Saved {target_name}: {dataset.shape}\")\n", "\n", "# Save metadata file\n", "metadata_file = os.path.join(dataset_dir, \"metadata.json\")\n", "with open(metadata_file, 'w') as f:\n", " json.dump(metadata, f, indent=2)\n", "\n", "print(f\"\\nāœ… All datasets saved:\")\n", "print(f\"šŸ“ Directory: {dataset_dir}/\")\n", "print(f\"šŸ“„ Files: {len(saved_files)} dataset files + 1 metadata file\")\n", "print(f\"šŸ“‹ Metadata: {metadata_file}\")\n", "\n", "# Display final directory contents\n", "print(f\"\\nšŸ“‚ Dataset directory contents:\")\n", "for file in os.listdir(dataset_dir):\n", " file_path = os.path.join(dataset_dir, file)\n", " if file.endswith('.pkl'):\n", " df_size = pd.read_pickle(file_path).shape\n", " print(f\" {file}: {df_size}\")\n", " else:\n", " print(f\" {file}\")\n" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.13.5" } }, "nbformat": 4, "nbformat_minor": 2 }