企业官网建设流程全解析

前言

在数据工程中，直接写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方言
分区 按列分区提高查询效率
桶 分桶优化JOIN
UDF 用户自定义函数

---

二、完整代码实现

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, orc
char delimiter;
struct table *next;
} table_t;

// 元数据存储（Metastore）
typedef 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, GROUPBY
char 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
// 创建Metastore
metastore_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_name+1, 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, "age>30") && i < 2) continue;
if (strstr(condition, "name='Alice'") && 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. 测试代码

```c
void 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;
}
```

---

三、编译和运行

```bash
gcc -o hive hive.c -lpthread
./hive
```

---

四、Hive vs 本实现

特性 本实现 Hive
SQL解析 ✅ 基础 ✅ 完整
元数据管理 ✅ ✅
分区表 ✅ ✅
执行引擎 ✅ 基础 ✅ MapReduce/Spark
查询优化 ❌ ✅
内置函数 ❌ ✅
UDF ❌ ✅

---

五、总结

通过这篇文章，你学会了：

· Hive的核心架构（Metastore + 解析器 + 执行引擎）
· 元数据管理（表结构、列、分区）
· SQL解析（SELECT、FROM、WHERE）
· 执行计划生成
· 数据扫描与过滤

Hive是数据仓库的经典实现。掌握它，你就理解了SQL-on-Hadoop的底层设计。

下一篇预告：《从零实现一个分布式调度：Apache Airflow的核心设计（进阶）》

---

评论区分享一下你用Hive处理过什么分析场景～