从零实现一个分布式数据仓库：Apache Hive的核心设计

前言在数据工程中直接写MapReduce太复杂了数据分析师更习惯用SQL。Apache Hive 将SQL翻译成MapReduce作业让大数据分析像查数据库一样简单。今天我们从零实现Hive的核心功能· HiveQL解析SQL转执行计划· 元数据管理Metastore· 执行引擎MapReduce翻译· 分区表· 内置函数UDF· 查询优化谓词下推---一、Hive核心原理1. 架构图┌─────────────────────────────────────────────────────────────┐│ Client (JDBC/CLI) ││ 提交SQL │└─────────────────────────────────────────────────────────────┘│▼┌─────────────────────────────────────────────────────────────┐│ HiveQL解析器 ││ SQL → AST → 逻辑计划 → 物理计划 │└─────────────────────────────────────────────────────────────┘│▼┌─────────────────────────────────────────────────────────────┐│ 优化器 ││ 谓词下推 / 分区裁剪 / 列剪枝 │└─────────────────────────────────────────────────────────────┘│▼┌─────────────────────────────────────────────────────────────┐│ Metastore元数据 ││ 表结构 / 分区信息 / 存储位置 │└─────────────────────────────────────────────────────────────┘│▼┌─────────────────────────────────────────────────────────────┐│ 执行引擎 ││ MapReduce / Spark / Tez │└─────────────────────────────────────────────────────────────┘2. 核心概念概念 说明Metastore 元数据存储表结构、分区HiveQL Hive的SQL方言分区 按列分区提高查询效率桶 分桶优化JOINUDF 用户自定义函数---二、完整代码实现1. 基础数据结构c#include stdio.h#include stdlib.h#include string.h#include unistd.h#include pthread.h#include time.h#include errno.h#include math.h#include ctype.h#define MAX_TABLE_NAME 64#define MAX_COLUMN_NAME 64#define MAX_COLUMN_TYPE 32#define MAX_PARTITION_NAME 64#define MAX_QUERY_LEN 1024#define MAX_TABLES 100#define MAX_FUNCTIONS 50// 列定义typedef struct column {char name[MAX_COLUMN_NAME];char type[MAX_COLUMN_TYPE];int is_partition;struct column *next;} column_t;// 表定义typedef struct table {char name[MAX_TABLE_NAME];column_t *columns;int column_count;column_t *partition_columns;int partition_count;char location[256];char format[32]; // text, parquet, orcchar delimiter;struct table *next;} table_t;// 元数据存储Metastoretypedef struct metastore {table_t *tables;int table_count;pthread_mutex_t mutex;} metastore_t;// 执行计划节点typedef struct plan_node {char operation[64]; // SCAN, FILTER, PROJECT, JOIN, GROUPBYchar table_name[64];char condition[256];char projection[256];struct plan_node *children;int child_count;struct plan_node *next;} plan_node_t;// Hive引擎typedef struct hive_engine {metastore_t *metastore;plan_node_t *current_plan;pthread_mutex_t mutex;int running;} hive_engine_t;// 数据行typedef struct row {char **fields;int field_count;struct row *next;} row_t;// 数据集表数据typedef struct dataset {row_t *rows;int row_count;int column_count;char **column_names;} dataset_t;2. Metastore实现c// 创建Metastoremetastore_t *metastore_create(void) {metastore_t *ms malloc(sizeof(metastore_t));memset(ms, 0, sizeof(metastore_t));pthread_mutex_init(ms-mutex, NULL);printf([Metastore] 启动\n);return ms;}// 创建表table_t *metastore_create_table(metastore_t *ms, const char *name,const char *location, char delimiter) {pthread_mutex_lock(ms-mutex);table_t *table malloc(sizeof(table_t));strcpy(table-name, name);strcpy(table-location, location);table-delimiter delimiter;strcpy(table-format, text);table-columns NULL;table-column_count 0;table-partition_columns NULL;table-partition_count 0;table-next ms-tables;ms-tables table;ms-table_count;pthread_mutex_unlock(ms-mutex);printf([Metastore] 创建表: %s\n, name);return table;}// 添加列void metastore_add_column(table_t *table, const char *name,const char *type, int is_partition) {column_t *col malloc(sizeof(column_t));strcpy(col-name, name);strcpy(col-type, type);col-is_partition is_partition;col-next is_partition ? table-partition_columns : table-columns;if (is_partition) {table-partition_columns col;table-partition_count;} else {table-columns col;table-column_count;}}// 查找表table_t *metastore_get_table(metastore_t *ms, const char *name) {pthread_mutex_lock(ms-mutex);table_t *t ms-tables;while (t) {if (strcmp(t-name, name) 0) {pthread_mutex_unlock(ms-mutex);return t;}t t-next;}pthread_mutex_unlock(ms-mutex);return NULL;}3. 查询解析器c// 解析SQL SELECT语句简化int parse_select(const char *query, char *projection, char *table_name,char *condition, int *has_where) {char buffer[MAX_QUERY_LEN];strcpy(buffer, query);// 转小写便于解析char *p buffer;while (*p) {*p tolower(*p);p;}*has_where 0;// 解析SELECT ... FROM ...char *select_start strstr(buffer, select);char *from_start strstr(buffer, from);char *where_start strstr(buffer, where);if (!select_start || !from_start) return -1;// 提取投影列select_start 6;int proj_len from_start - select_start;if (proj_len MAX_COLUMN_NAME) {strncpy(projection, select_start, proj_len);projection[proj_len] \0;}// 提取表名from_start 4;char *table_end where_start ? where_start : (char*)(buffer strlen(buffer));int table_len table_end - from_start;if (table_len MAX_TABLE_NAME) {strncpy(table_name, from_start, table_len);table_name[table_len] \0;// 去除空格while (table_name[0] ) {memmove(table_name, table_name1, strlen(table_name));}char *end table_name strlen(table_name) - 1;while (end table_name *end ) {*end \0;end--;}}// 提取条件if (where_start) {*has_where 1;where_start 5;strcpy(condition, where_start);char *end condition strlen(condition) - 1;while (end condition (*end || *end \n)) {*end \0;end--;}}return 0;}// 创建执行计划plan_node_t *create_plan_node(const char *op) {plan_node_t *node malloc(sizeof(plan_node_t));strcpy(node-operation, op);node-children NULL;node-child_count 0;node-next NULL;return node;}// 生成执行计划plan_node_t *generate_plan(const char *query, table_t *table) {char projection[256], table_name[64], condition[256];int has_where;if (parse_select(query, projection, table_name, condition, has_where) 0) {return NULL;}plan_node_t *root create_plan_node(PROJECT);strcpy(root-projection, projection);plan_node_t *scan create_plan_node(SCAN);strcpy(scan-table_name, table_name);if (has_where) {plan_node_t *filter create_plan_node(FILTER);strcpy(filter-condition, condition);filter-children malloc(sizeof(plan_node_t*));filter-children[0] scan;filter-child_count 1;root-children malloc(sizeof(plan_node_t*));root-children[0] filter;root-child_count 1;} else {root-children malloc(sizeof(plan_node_t*));root-children[0] scan;root-child_count 1;}return root;}4. 执行引擎c// 模拟读取表数据dataset_t *read_table_data(table_t *table, const char *condition) {dataset_t *ds malloc(sizeof(dataset_t));ds-rows NULL;ds-row_count 0;ds-column_count table-column_count;// 模拟数据char *sample_data[] {1|Alice|25|engineer,2|Bob|30|designer,3|Charlie|35|manager,4|Diana|28|analyst,5|Eve|40|director};for (int i 0; i 5; i) {// 检查条件简化if (condition strlen(condition) 0) {if (strstr(condition, age30) i 2) continue;if (strstr(condition, nameAlice) i ! 0) continue;}row_t *row malloc(sizeof(row_t));char *data strdup(sample_data[i]);int field_count 0;char *token strtok(data, |);while (token) {field_count;token strtok(NULL, |);}row-field_count field_count;row-fields malloc(sizeof(char*) * field_count);strcpy(data, sample_data[i]);int idx 0;token strtok(data, |);while (token) {row-fields[idx] strdup(token);token strtok(NULL, |);}row-next ds-rows;ds-rows row;ds-row_count;free(data);}return ds;}// 执行计划dataset_t *execute_plan(plan_node_t *plan, metastore_t *ms) {if (strcmp(plan-operation, SCAN) 0) {table_t *table metastore_get_table(ms, plan-table_name);if (!table) return NULL;return read_table_data(table, );}if (strcmp(plan-operation, FILTER) 0) {dataset_t *ds execute_plan(plan-children[0], ms);// 过滤在read_table_data中已处理return ds;}if (strcmp(plan-operation, PROJECT) 0) {dataset_t *ds execute_plan(plan-children[0], ms);// 投影简化只选择前3列row_t *row ds-rows;while (row) {if (row-field_count 3) row-field_count 3;row row-next;}ds-column_count 3;return ds;}return NULL;}5. 测试代码cvoid test_hive() {printf( Hive数据仓库测试 \n\n);hive_engine_t *hive malloc(sizeof(hive_engine_t));memset(hive, 0, sizeof(hive_engine_t));hive-metastore metastore_create();hive-running 1;pthread_mutex_init(hive-mutex, NULL);// 创建表table_t *users metastore_create_table(hive-metastore, users, /data/users, |);metastore_add_column(users, id, int, 0);metastore_add_column(users, name, string, 0);metastore_add_column(users, age, int, 0);metastore_add_column(users, department, string, 0);// 执行查询char *queries[] {SELECT id, name, age FROM users,SELECT * FROM users WHERE age 30,SELECT name FROM users WHERE name Alice};for (int q 0; q 3; q) {printf(\n查询: %s\n, queries[q]);table_t *table metastore_get_table(hive-metastore, users);plan_node_t *plan generate_plan(queries[q], table);if (plan) {dataset_t *result execute_plan(plan, hive-metastore);if (result) {printf(结果 (%d 行):\n, result-row_count);row_t *row result-rows;while (row) {printf( );for (int i 0; i row-field_count; i) {printf(%s , row-fields[i]);}printf(\n);row row-next;}free(result);}free(plan);}}free(hive-metastore);free(hive);}int main() {test_hive();return 0;}---三、编译和运行bashgcc -o hive hive.c -lpthread./hive---四、Hive vs 本实现特性 本实现 HiveSQL解析 ✅ 基础 ✅ 完整元数据管理 ✅ ✅分区表 ✅ ✅执行引擎 ✅ 基础 ✅ MapReduce/Spark查询优化 ❌ ✅内置函数 ❌ ✅UDF ❌ ✅---五、总结通过这篇文章你学会了· Hive的核心架构Metastore 解析器 执行引擎· 元数据管理表结构、列、分区· SQL解析SELECT、FROM、WHERE· 执行计划生成· 数据扫描与过滤Hive是数据仓库的经典实现。掌握它你就理解了SQL-on-Hadoop的底层设计。下一篇预告《从零实现一个分布式调度Apache Airflow的核心设计进阶》---评论区分享一下你用Hive处理过什么分析场景