import heapq
from collections import defaultdict

class HuffmanNode:
    def __init__(self, char=None, freq=0, left=None, right=None):
        self.char = char
        self.freq = freq
        self.left = left
        self.right = right
    
    def __lt__(self, other):
        return self.freq < other.freq

def build_huffman_tree(text):
    # 统计字符频率
    frequency = defaultdict(int)
    for char in text:
        frequency[char] += 1
    
    # 创建叶子节点
    heap = []
    for char, freq in frequency.items():
        heapq.heappush(heap, HuffmanNode(char=char, freq=freq))
    
    # 构建霍夫曼树
    while len(heap) > 1:
        left = heapq.heappop(heap)
        right = heapq.heappop(heap)
        parent = HuffmanNode(freq=left.freq + right.freq, left=left, right=right)
        heapq.heappush(heap, parent)
    
    return heap[0]

def generate_codes(root, current_code="", codes=None):
    if codes is None:
        codes = {}
    
    if root is None:
        return
    
    if root.char is not None:
        codes[root.char] = current_code
        return
    
    generate_codes(root.left, current_code + "0", codes)
    generate_codes(root.right, current_code + "1", codes)
    
    return codes

def huffman_encode(text):
    root = build_huffman_tree(text)
    codes = generate_codes(root)
    
    encoded_text = ""
    for char in text:
        encoded_text += codes[char]
    
    return encoded_text, codes

# 示例使用
text = "this is an example for huffman encoding"
encoded_text, codes = huffman_encode(text)
print("编码结果:", encoded_text)
print("编码表:", codes)

-- 创建表
CREATE TABLE customers (
    customer_id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    phone VARCHAR(20),
    address VARCHAR(200)
);

-- 插入数据
INSERT INTO customers (customer_id, name, email, phone, address)
VALUES (1, '张三', 'zhangsan@example.com', '13800138000', '北京市朝阳区');

-- 查询数据
SELECT * FROM customers WHERE name = '张三';

-- 创建索引以提高查询性能
CREATE INDEX idx_customer_name ON customers(name);

// 连接到MongoDB
const { MongoClient } = require('mongodb');
const uri = "mongodb://localhost:27017";
const client = new MongoClient(uri);

async function run() {
    try {
        await client.connect();
        const database = client.db("mydb");
        const customers = database.collection("customers");
        
        // 插入文档
        const doc = {
            name: "张三",
            email: "zhangsan@example.com",
            phone: "13800138000",
            address: {
                city: "北京",
                district: "朝阳区"
            },
            orders: [
                { order_id: "ORD001", amount: 100 },
                { order_id: "ORD002", amount: 200 }
            ]
        };
        const result = await customers.insertOne(doc);
        console.log(`插入文档ID: ${result.insertedId}`);
        
        // 查询文档
        const query = { name: "张三" };
        const customer = await customers.findOne(query);
        console.log(customer);
        
        // 创建索引
        const indexResult = await customers.createIndex({ name: 1 });
        console.log(`创建索引: ${indexResult}`);
    } finally {
        await client.close();
    }
}

run().catch(console.dir);

-- 创建事实表
CREATE TABLE sales_fact (
    sale_id INT PRIMARY KEY,
    date_id INT,
    product_id INT,
    customer_id INT,
    store_id INT,
    quantity INT,
    amount DECIMAL(10,2)
);

-- 创建维度表
CREATE TABLE date_dim (
    date_id INT PRIMARY KEY,
    date DATE,
    day INT,
    month INT,
    quarter INT,
    year INT
);

CREATE TABLE product_dim (
    product_id INT PRIMARY KEY,
    product_name VARCHAR(100),
    category VARCHAR(50),
    brand VARCHAR(50),
    price DECIMAL(10,2)
);

CREATE TABLE customer_dim (
    customer_id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    gender CHAR(1),
    age INT,
    city VARCHAR(50),
    country VARCHAR(50)
);

CREATE TABLE store_dim (
    store_id INT PRIMARY KEY,
    store_name VARCHAR(100),
    city VARCHAR(50),
    country VARCHAR(50),
    manager VARCHAR(100)
);

-- 查询示例：按季度和产品类别分析销售额
SELECT 
    d.quarter,
    p.category,
    SUM(s.amount) AS total_sales,
    SUM(s.quantity) AS total_quantity
FROM 
    sales_fact s
JOIN date_dim d ON s.date_id = d.date_id
JOIN product_dim p ON s.product_id = p.product_id
GROUP BY 
    d.quarter, p.category
ORDER BY 
    d.quarter, total_sales DESC;

import boto3

# 创建S3客户端
s3 = boto3.client('s3')

# 创建存储桶
bucket_name = 'my-big-data-bucket'
s3.create_bucket(Bucket=bucket_name)

# 上传文件
file_name = 'large_dataset.csv'
s3.upload_file(file_name, bucket_name, file_name)

# 列出存储桶中的对象
response = s3.list_objects_v2(Bucket=bucket_name)
for obj in response.get('Contents', []):
    print(f"对象: {obj['Key']}, 大小: {obj['Size']} 字节")

# 下载文件
s3.download_file(bucket_name, file_name, 'downloaded_' + file_name)

# 删除对象
s3.delete_object(Bucket=bucket_name, Key=file_name)

# 删除存储桶
s3.delete_bucket(Bucket=bucket_name)

from minio import Minio
from minio.error import S3Error

# 创建MinIO客户端
client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True
)

# 创建存储桶
bucket_name = "my-big-data-bucket"
try:
    client.make_bucket(bucket_name)
except S3Error as err:
    print(err)

# 上传对象
object_name = "large_dataset.csv"
file_path = "./" + object_name
try:
    client.fput_object(bucket_name, object_name, file_path)
    print(f"成功上传 {object_name}")
except S3Error as err:
    print(err)

# 列出存储桶中的对象
try:
    objects = client.list_objects(bucket_name)
    for obj in objects:
        print(f"对象: {obj.object_name}, 大小: {obj.size} 字节")
except S3Error as err:
    print(err)

# 下载对象
try:
    client.fget_object(bucket_name, object_name, "downloaded_" + object_name)
    print(f"成功下载 {object_name}")
except S3Error as err:
    print(err)

# 删除对象
try:
    client.remove_object(bucket_name, object_name)
    print(f"成功删除 {object_name}")
except S3Error as err:
    print(err)

# 删除存储桶
try:
    client.remove_bucket(bucket_name)
    print(f"成功删除存储桶 {bucket_name}")
except S3Error as err:
    print(err)

class BTreeNode:
    def __init__(self, leaf=False):
        self.leaf = leaf  # 是否为叶子节点
        self.keys = []    # 键值列表
        self.children = []  # 子节点列表

class BTree:
    def __init__(self, t):
        self.root = BTreeNode(True)  # 根节点
        self.t = t  # 最小度数，决定节点中键的最小和最大数量
    
    def search(self, k, node=None):
        # 递归搜索键k
        if node is None:
            node = self.root
        
        i = 0
        while i < len(node.keys) and k > node.keys[i]:
            i += 1
        
        if i < len(node.keys) and k == node.keys[i]:
            return (node, i)  # 找到键，返回节点和位置
        
        if node.leaf:
            return None  # 未找到键
        
        return self.search(k, node.children[i])
    
    def insert(self, k):
        root = self.root
        if len(root.keys) == (2 * self.t) - 1:
            # 根节点已满，创建新根
            new_root = BTreeNode()
            new_root.children.append(self.root)
            self.root = new_root
            self._split_child(new_root, 0)
            self._insert_nonfull(new_root, k)
        else:
            self._insert_nonfull(root, k)
    
    def _split_child(self, parent, i):
        # 分裂父节点的第i个子节点
        t = self.t
        y = parent.children[i]
        z = BTreeNode(y.leaf)
        
        # 将y的后t-1个键移动到z
        z.keys = y.keys[t:]
        y.keys = y.keys[:t-1]
        
        # 如果y不是叶子节点，将y的后t个子节点移动到z
        if not y.leaf:
            z.children = y.children[t:]
            y.children = y.children[:t]
        
        # 将z插入父节点
        parent.children.insert(i+1, z)
        parent.keys.insert(i, y.keys[t-1])
    
    def _insert_nonfull(self, node, k):
        i = len(node.keys) - 1
        
        if node.leaf:
            # 在叶子节点中插入键
            node.keys.append(0)
            while i >= 0 and k < node.keys[i]:
                node.keys[i+1] = node.keys[i]
                i -= 1
            node.keys[i+1] = k
        else:
            # 找到合适的子节点
            while i >= 0 and k < node.keys[i]:
                i -= 1
            i += 1
            
            # 如果子节点已满，先分裂
            if len(node.children[i].keys) == (2 * self.t) - 1:
                self._split_child(node, i)
                if k > node.keys[i]:
                    i += 1
            
            self._insert_nonfull(node.children[i], k)
    
    def delete(self, k):
        self._delete(self.root, k)
    
    def _delete(self, node, k):
        t = self.t
        i = 0
        while i < len(node.keys) and k > node.keys[i]:
            i += 1
        
        if i < len(node.keys) and k == node.keys[i]:
            # 情况1：键在当前节点中
            if node.leaf:
                # 情况1a：从叶子节点中删除
                node.keys.pop(i)
            else:
                # 情况1b：从内部节点中删除
                if len(node.children[i].keys) >= t:
                    # 情况1b-i：前驱子节点至少有t个键
                    predecessor = self._get_predecessor(node.children[i])
                    node.keys[i] = predecessor
                    self._delete(node.children[i], predecessor)
                elif len(node.children[i+1].keys) >= t:
                    # 情况1b-ii：后继子节点至少有t个键
                    successor = self._get_successor(node.children[i+1])
                    node.keys[i] = successor
                    self._delete(node.children[i+1], successor)
                else:
                    # 情况1b-iii：合并两个子节点
                    self._merge_children(node, i)
                    self._delete(node.children[i], k)
        else:
            # 情况2：键不在当前节点中
            if node.leaf:
                # 键不存在于树中
                return
            
            # 确保要访问的子节点至少有t个键
            flag = (i == len(node.keys))
            if len(node.children[i].keys) < t:
                self._fill(node, i)
            
            if flag and i > len(node.keys):
                self._delete(node.children[i-1], k)
            else:
                self._delete(node.children[i], k)
    
    def _get_predecessor(self, node):
        # 获取节点中的最大键
        while not node.leaf:
            node = node.children[-1]
        return node.keys[-1]
    
    def _get_successor(self, node):
        # 获取节点中的最小键
        while not node.leaf:
            node = node.children[0]
        return node.keys[0]
    
    def _merge_children(self, node, i):
        # 合并节点的第i和第i+1个子节点
        t = self.t
        child = node.children[i]
        next_child = node.children[i+1]
        
        # 将父节点的键下移到子节点
        child.keys.append(node.keys[i])
        
        # 将next_child的键和子节点合并到child
        child.keys.extend(next_child.keys)
        if not child.leaf:
            child.children.extend(next_child.children)
        
        # 从父节点中删除键和next_child
        node.keys.pop(i)
        node.children.pop(i+1)
    
    def _fill(self, node, i):
        t = self.t
        if i > 0 and len(node.children[i-1].keys) >= t:
            # 从前一个兄弟节点借用一个键
            self._borrow_from_prev(node, i)
        elif i < len(node.children) and len(node.children[i+1].keys) >= t:
            # 从后一个兄弟节点借用一个键
            self._borrow_from_next(node, i)
        else:
            # 合并子节点
            if i < len(node.children):
                self._merge_children(node, i)
            else:
                self._merge_children(node, i-1)
    
    def _borrow_from_prev(self, node, i):
        child = node.children[i]
        sibling = node.children[i-1]
        
        # 将父节点的键下移到子节点
        child.keys.insert(0, node.keys[i-1])
        
        # 如果不是叶子节点，将兄弟节点的最后一个子节点移动到子节点
        if not child.leaf:
            child.children.insert(0, sibling.children.pop())
        
        # 将兄弟节点的最后一个键上移到父节点
        node.keys[i-1] = sibling.keys.pop()
    
    def _borrow_from_next(self, node, i):
        child = node.children[i]
        sibling = node.children[i+1]
        
        # 将父节点的键下移到子节点
        child.keys.append(node.keys[i])
        
        # 如果不是叶子节点，将兄弟节点的第一个子节点移动到子节点
        if not child.leaf:
            child.children.append(sibling.children.pop(0))
        
        # 将兄弟节点的第一个键上移到父节点
        node.keys[i] = sibling.keys.pop(0)

# 示例使用
btree = BTree(3)  # 最小度数为3的B树
keys = [10, 20, 5, 6, 12, 30, 7, 17]

for key in keys:
    btree.insert(key)

# 搜索键
result = btree.search(12)
if result:
    print(f"找到键12在节点中")
else:
    print("未找到键12")

# 删除键
btree.delete(12)
result = btree.search(12)
if result:
    print(f"找到键12在节点中")
else:
    print("未找到键12")

class HashIndex:
    def __init__(self, size=10):
        self.size = size
        self.buckets = [[] for _ in range(size)]  # 哈希桶列表
    
    def _hash_function(self, key):
        # 简单的哈希函数
        return hash(key) % self.size
    
    def insert(self, key, value):
        # 插入键值对
        hash_value = self._hash_function(key)
        bucket = self.buckets[hash_value]
        
        # 检查键是否已存在
        for i, (k, v) in enumerate(bucket):
            if k == key:
                bucket[i] = (key, value)  # 更新值
                return
        
        # 键不存在，添加新键值对
        bucket.append((key, value))
    
    def search(self, key):
        # 搜索键
        hash_value = self._hash_function(key)
        bucket = self.buckets[hash_value]
        
        for k, v in bucket:
            if k == key:
                return v  # 返回值
        
        return None  # 键不存在
    
    def delete(self, key):
        # 删除键
        hash_value = self._hash_function(key)
        bucket = self.buckets[hash_value]
        
        for i, (k, v) in enumerate(bucket):
            if k == key:
                bucket.pop(i)  # 删除键值对
                return True
        
        return False  # 键不存在

# 示例使用
index = HashIndex()

# 插入数据
index.insert("name", "张三")
index.insert("age", 30)
index.insert("city", "北京")

# 搜索数据
print(index.search("name"))  # 输出: 张三
print(index.search("age"))   # 输出: 30
print(index.search("city"))  # 输出: 北京
print(index.search("country"))  # 输出: None

# 删除数据
index.delete("age")
print(index.search("age"))   # 输出: None

class BitmapIndex:
    def __init__(self):
        self.values = {}  # 值到位图的映射
        self.records = []  # 记录列表
    
    def add_record(self, record_id, value):
        # 添加记录
        if record_id >= len(self.records):
            # 扩展记录列表
            self.records.extend([None] * (record_id - len(self.records) + 1))
        
        self.records[record_id] = value
        
        # 更新位图
        if value not in self.values:
            self.values[value] = [0] * len(self.records)
        
        # 设置对应位为1
        self.values[value][record_id] = 1
    
    def search(self, value):
        # 搜索具有特定值的记录
        if value not in self.values:
            return []
        
        bitmap = self.values[value]
        return [i for i, bit in enumerate(bitmap) if bit == 1]
    
    def search_and(self, value1, value2):
        # 搜索同时具有两个值的记录（AND操作）
        if value1 not in self.values or value2 not in self.values:
            return []
        
        bitmap1 = self.values[value1]
        bitmap2 = self.values[value2]
        
        result = []
        for i in range(len(bitmap1)):
            if bitmap1[i] == 1 and bitmap2[i] == 1:
                result.append(i)
        
        return result
    
    def search_or(self, value1, value2):
        # 搜索具有任一值的记录（OR操作）
        if value1 not in self.values and value2 not in self.values:
            return []
        elif value1 not in self.values:
            return self.search(value2)
        elif value2 not in self.values:
            return self.search(value1)
        
        bitmap1 = self.values[value1]
        bitmap2 = self.values[value2]
        
        result = []
        for i in range(len(bitmap1)):
            if bitmap1[i] == 1 or bitmap2[i] == 1:
                result.append(i)
        
        return result
    
    def search_not(self, value):
        # 搜索不具有特定值的记录（NOT操作）
        if value not in self.values:
            return list(range(len(self.records)))
        
        bitmap = self.values[value]
        return [i for i, bit in enumerate(bitmap) if bit == 0 and self.records[i] is not None]

# 示例使用
index = BitmapIndex()

# 添加记录
index.add_record(0, "红色")
index.add_record(1, "蓝色")
index.add_record(2, "红色")
index.add_record(3, "绿色")
index.add_record(4, "蓝色")
index.add_record(5, "红色")

# 搜索记录
print("红色的记录:", index.search("红色"))  # 输出: [0, 2, 5]
print("蓝色的记录:", index.search("蓝色"))  # 输出: [1, 4]
print("绿色的记录:", index.search("绿色"))  # 输出: [3]

# 复合查询
print("红色或蓝色的记录:", index.search_or("红色", "蓝色"))  # 输出: [0, 1, 2, 4, 5]
print("不是红色的记录:", index.search_not("红色"))  # 输出: [1, 3, 4]

import hashlib
import os

class DataChunker:
    def __init__(self, chunk_size=1024*1024):  # 默认块大小为1MB
        self.chunk_size = chunk_size
    
    def fixed_size_chunking(self, data):
        """固定大小分块"""
        chunks = []
        for i in range(0, len(data), self.chunk_size):
            chunk = data[i:i+self.chunk_size]
            chunk_id = i // self.chunk_size
            chunks.append({
                'id': chunk_id,
                'data': chunk,
                'size': len(chunk),
                'hash': hashlib.md5(chunk).hexdigest()
            })
        return chunks
    
    def content_based_chunking(self, data, pattern=b'\n'):
        """基于内容分块（按行分割）"""
        chunks = []
        lines = data.split(pattern)
        current_chunk = b''
        chunk_id = 0
        
        for line in lines:
            if len(current_chunk) + len(line) + len(pattern) > self.chunk_size and current_chunk:
                # 当前块已满，保存并开始新块
                chunks.append({
                    'id': chunk_id,
                    'data': current_chunk,
                    'size': len(current_chunk),
                    'hash': hashlib.md5(current_chunk).hexdigest()
                })
                chunk_id += 1
                current_chunk = line + pattern
            else:
                current_chunk += line + pattern
        
        # 添加最后一个块
        if current_chunk:
            chunks.append({
                'id': chunk_id,
                'data': current_chunk,
                'size': len(current_chunk),
                'hash': hashlib.md5(current_chunk).hexdigest()
            })
        
        return chunks
    
    def hash_based_chunking(self, data, window_size=48, chunk_size_target=1024*1024):
        """基于哈希分块（用于去重）"""
        chunks = []
        start = 0
        chunk_id = 0
        
        while start < len(data):
            # 查找下一个断点
            end = min(start + chunk_size_target, len(data))
            if end < len(data):
                # 在窗口内查找哈希值满足条件的断点
                for i in range(end - window_size, end):
                    window = data[i:i+window_size]
                    hash_value = int(hashlib.md5(window).hexdigest(), 16)
                    if hash_value % 1024 == 0:  # 使用哈希值的低10位作为条件
                        end = i + window_size
                        break
            
            # 创建块
            chunk = data[start:end]
            chunks.append({
                'id': chunk_id,
                'data': chunk,
                'size': len(chunk),
                'hash': hashlib.md5(chunk).hexdigest()
            })
            
            start = end
            chunk_id += 1
        
        return chunks
    
    def range_based_chunking(self, data, key_func, num_chunks=10):
        """基于范围分块"""
        # 提取键并排序
        items = [(key_func(item), item) for item in data]
        items.sort(key=lambda x: x[0])
        
        # 计算每个块的范围
        keys = [item[0] for item in items]
        min_key, max_key = min(keys), max(keys)
        range_size = (max_key - min_key) / num_chunks
        
        # 分割数据
        chunks = []
        current_chunk = []
        current_range = min_key + range_size
        chunk_id = 0
        
        for key, item in items:
            if key > current_range and current_chunk:
                # 当前范围已满，保存并开始新块
                chunks.append({
                    'id': chunk_id,
                    'data': current_chunk,
                    'size': len(current_chunk),
                    'range': (current_range - range_size, current_range)
                })
                chunk_id += 1
                current_chunk = []
                current_range += range_size
            
            current_chunk.append(item)
        
        # 添加最后一个块
        if current_chunk:
            chunks.append({
                'id': chunk_id,
                'data': current_chunk,
                'size': len(current_chunk),
                'range': (current_range - range_size, current_range)
            })
        
        return chunks
    
    def save_chunks(self, chunks, output_dir):
        """保存块到文件"""
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
        
        metadata = []
        for chunk in chunks:
            # 保存块数据
            chunk_filename = os.path.join(output_dir, f"chunk_{chunk['id']}.dat")
            with open(chunk_filename, 'wb') as f:
                f.write(chunk['data'])
            
            # 记录元数据
            metadata.append({
                'id': chunk['id'],
                'filename': chunk_filename,
                'size': chunk['size'],
                'hash': chunk['hash']
            })
        
        # 保存元数据
        metadata_filename = os.path.join(output_dir, "metadata.json")
        import json
        with open(metadata_filename, 'w') as f:
            json.dump(metadata, f)
        
        return metadata_filename
    
    def load_chunks(self, metadata_filename):
        """从文件加载块"""
        import json
        with open(metadata_filename, 'r') as f:
            metadata = json.load(f)
        
        chunks = []
        for meta in metadata:
            with open(meta['filename'], 'rb') as f:
                data = f.read()
            
            # 验证哈希
            if hashlib.md5(data).hexdigest() != meta['hash']:
                raise ValueError(f"块 {meta['id']} 的哈希值不匹配")
            
            chunks.append({
                'id': meta['id'],
                'data': data,
                'size': meta['size'],
                'hash': meta['hash']
            })
        
        return chunks
    
    def reassemble_data(self, chunks):
        """重组数据"""
        # 按ID排序块
        sorted_chunks = sorted(chunks, key=lambda x: x['id'])
        
        # 连接数据
        data = b''
        for chunk in sorted_chunks:
            data += chunk['data']
        
        return data

# 示例使用
chunker = DataChunker(chunk_size=1024)  # 1KB块大小

# 生成测试数据
test_data = b"这是一个测试数据，用于演示数据分块技术。" * 100

# 固定大小分块
fixed_chunks = chunker.fixed_size_chunking(test_data)
print(f"固定大小分块: {len(fixed_chunks)} 个块")

# 基于内容分块（按行）
content_chunks = chunker.content_based_chunking(test_data, pattern=b'。')
print(f"基于内容分块: {len(content_chunks)} 个块")

# 基于哈希分块
hash_chunks = chunker.hash_based_chunking(test_data, chunk_size_target=512)
print(f"基于哈希分块: {len(hash_chunks)} 个块")

# 基于范围分块（假设数据是数字列表）
numeric_data = list(range(100))
range_chunks = chunker.range_based_chunking(numeric_data, key_func=lambda x: x, num_chunks=5)
print(f"基于范围分块: {len(range_chunks)} 个块")

# 保存和加载块
output_dir = "chunks"
metadata_file = chunker.save_chunks(fixed_chunks, output_dir)
loaded_chunks = chunker.load_chunks(metadata_file)

# 重组数据
reassembled_data = chunker.reassemble_data(loaded_chunks)
print(f"重组数据成功: {reassembled_data == test_data}")

-- 创建患者表
CREATE TABLE patients (
    patient_id INT PRIMARY KEY,
    name VARCHAR(100),
    date_of_birth DATE,
    gender CHAR(1),
    blood_type VARCHAR(5),
    contact_number VARCHAR(20),
    address VARCHAR(200)
);

-- 创建诊断表
CREATE TABLE diagnoses (
    diagnosis_id INT PRIMARY KEY,
    patient_id INT,
    diagnosis_date DATE,
    diagnosis_code VARCHAR(20),
    description TEXT,
    doctor_id INT,
    FOREIGN KEY (patient_id) REFERENCES patients(patient_id)
);

-- 创建治疗表
CREATE TABLE treatments (
    treatment_id INT PRIMARY KEY,
    patient_id INT,
    diagnosis_id INT,
    treatment_date DATE,
    treatment_type VARCHAR(50),
    medication VARCHAR(100),
    dosage VARCHAR(50),
    notes TEXT,
    FOREIGN KEY (patient_id) REFERENCES patients(patient_id),
    FOREIGN KEY (diagnosis_id) REFERENCES diagnoses(diagnosis_id)
);

-- 创建医学影像表
CREATE TABLE medical_images (
    image_id INT PRIMARY KEY,
    patient_id INT,
    image_date DATE,
    image_type VARCHAR(50),
    image_path VARCHAR(200),
    description TEXT,
    radiologist_id INT,
    FOREIGN KEY (patient_id) REFERENCES patients(patient_id)
);

-- 查询示例：获取糖尿病患者的治疗历史
SELECT 
    p.patient_id,
    p.name,
    d.diagnosis_date,
    d.description AS diagnosis,
    t.treatment_date,
    t.treatment_type,
    t.medication,
    t.dosage
FROM 
    patients p
JOIN diagnoses d ON p.patient_id = d.patient_id
LEFT JOIN treatments t ON d.diagnosis_id = t.diagnosis_id
WHERE 
    d.diagnosis_code = 'E11'  -- E11是糖尿病的ICD-10代码
ORDER BY 
    p.patient_id, d.diagnosis_date, t.treatment_date;

-- 创建索引以提高查询性能
CREATE INDEX idx_diagnoses_patient_id ON diagnoses(patient_id);
CREATE INDEX idx_diagnoses_code ON diagnoses(diagnosis_code);
CREATE INDEX idx_treatments_patient_id ON treatments(patient_id);
CREATE INDEX idx_treatments_diagnosis_id ON treatments(diagnosis_id);

import boto3
import pandas as pd
from io import StringIO

# 创建S3和Athena客户端
s3 = boto3.client('s3')
athena = boto3.client('athena')

# 定义参数
bucket_name = 'my-big-data-bucket'
database_name = 'my_database'
table_name = 'sales_data'
query_output = 's3://my-big-data-bucket/query-results/'

# 上传CSV数据到S3
csv_data = """date,product,category,price,quantity
2023-01-01,Laptop,Electronics,999.99,5
2023-01-02,Phone,Electronics,699.99,10
2023-01-03,Shirt,Clothing,29.99,20
2023-01-04,Book,Education,15.99,30
2023-01-05,Headphones,Electronics,99.99,15"""

s3.put_object(Bucket=bucket_name, Key='sales/sales_data.csv', Body=csv_data)

# 创建Athena数据库
try:
    athena.create_database(Name=database_name)
    print(f"数据库 {database_name} 创建成功")
except athena.exceptions.AlreadyExistsException:
    print(f"数据库 {database_name} 已存在")

# 创建Athena表
create_table_query = f"""
CREATE EXTERNAL TABLE IF NOT EXISTS {database_name}.{table_name} (
    `date` DATE,
    product STRING,
    category STRING,
    price DOUBLE,
    quantity INT
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 's3://{bucket_name}/sales/'
TBLPROPERTIES ('skip.header.line.count'='1')
"""

response = athena.start_query_execution(
    QueryString=create_table_query,
    QueryExecutionContext={'Database': database_name},
    ResultConfiguration={'OutputLocation': query_output}
)

# 等待查询完成
execution_id = response['QueryExecutionId']
status = 'RUNNING'
while status in ['RUNNING', 'QUEUED']:
    response = athena.get_query_execution(QueryExecutionId=execution_id)
    status = response['QueryExecution']['Status']['State']
    
    if status == 'FAILED':
        raise Exception(f"查询失败: {response['QueryExecution']['Status']['StateChangeReason']}")
    elif status == 'SUCCEEDED':
        print("表创建成功")
        break
    else:
        import time
        time.sleep(1)

# 执行查询分析数据
query = f"""
SELECT 
    category,
    SUM(price * quantity) AS total_revenue,
    SUM(quantity) AS total_quantity,
    AVG(price) AS avg_price
FROM {database_name}.{table_name}
GROUP BY category
ORDER BY total_revenue DESC
"""

response = athena.start_query_execution(
    QueryString=query,
    QueryExecutionContext={'Database': database_name},
    ResultConfiguration={'OutputLocation': query_output}
)

# 等待查询完成
execution_id = response['QueryExecutionId']
status = 'RUNNING'
while status in ['RUNNING', 'QUEUED']:
    response = athena.get_query_execution(QueryExecutionId=execution_id)
    status = response['QueryExecution']['Status']['State']
    
    if status == 'FAILED':
        raise Exception(f"查询失败: {response['QueryExecution']['Status']['StateChangeReason']}")
    elif status == 'SUCCEEDED':
        print("查询执行成功")
        break
    else:
        import time
        time.sleep(1)

# 获取查询结果
result_location = response['QueryExecution']['ResultConfiguration']['OutputLocation']
result_key = result_location.replace(f's3://{bucket_name}/', '')
result_obj = s3.get_object(Bucket=bucket_name, Key=result_key)
result_data = result_obj['Body'].read().decode('utf-8')

# 将结果转换为DataFrame
df = pd.read_csv(StringIO(result_data))
print("分析结果:")
print(df)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, avg, count
import boto3

# 创建Spark会话
spark = SparkSession.builder \
    .appName("BigDataAnalysis") \
    .getOrCreate()

# 从S3读取数据
input_path = "s3://my-big-data-bucket/sales/"
sales_df = spark.read.csv(input_path, header=True, inferSchema=True)

# 数据清洗和转换
# 过滤掉无效记录
cleaned_df = sales_df.filter(
    (col("price") > 0) & 
    (col("quantity") > 0) & 
    (col("date").isNotNull())
)

# 添加计算字段
transformed_df = cleaned_df.withColumn("total_amount", col("price") * col("quantity"))

# 数据分析
# 按类别分析销售情况
category_analysis = transformed_df.groupBy("category").agg(
    sum("total_amount").alias("total_revenue"),
    sum("quantity").alias("total_quantity"),
    avg("price").alias("avg_price"),
    count("product").alias("product_count")
).orderBy(col("total_revenue").desc())

# 按产品分析销售情况
product_analysis = transformed_df.groupBy("product").agg(
    sum("total_amount").alias("total_revenue"),
    sum("quantity").alias("total_quantity")
).orderBy(col("total_revenue").desc()).limit(10)

# 按日期分析销售趋势
date_analysis = transformed_df.groupBy("date").agg(
    sum("total_amount").alias("daily_revenue"),
    sum("quantity").alias("daily_quantity")
).orderBy("date")

# 将分析结果保存回S3
output_path = "s3://my-big-data-bucket/analysis-results/"

category_analysis.write.mode("overwrite").csv(f"{output_path}category_analysis", header=True)
product_analysis.write.mode("overwrite").csv(f"{output_path}product_analysis", header=True)
date_analysis.write.mode("overwrite").csv(f"{output_path}date_analysis", header=True)

# 使用AWS Glue Catalog创建表
glue = boto3.client('glue')

# 创建数据库
try:
    glue.create_database(
        DatabaseInput={
            'Name': 'sales_analysis_db',
            'Description': 'Database for sales analysis results'
        }
    )
    print("数据库创建成功")
except glue.exceptions.AlreadyExistsException:
    print("数据库已存在")

# 创建表
tables = [
    {
        'name': 'category_analysis',
        'path': f"{output_path}category_analysis/",
        'columns': [
            {'name': 'category', 'type': 'string'},
            {'name': 'total_revenue', 'type': 'double'},
            {'name': 'total_quantity', 'type': 'bigint'},
            {'name': 'avg_price', 'type': 'double'},
            {'name': 'product_count', 'type': 'bigint'}
        ]
    },
    {
        'name': 'product_analysis',
        'path': f"{output_path}product_analysis/",
        'columns': [
            {'name': 'product', 'type': 'string'},
            {'name': 'total_revenue', 'type': 'double'},
            {'name': 'total_quantity', 'type': 'bigint'}
        ]
    },
    {
        'name': 'date_analysis',
        'path': f"{output_path}date_analysis/",
        'columns': [
            {'name': 'date', 'type': 'date'},
            {'name': 'daily_revenue', 'type': 'double'},
            {'name': 'daily_quantity', 'type': 'bigint'}
        ]
    }
]

for table in tables:
    try:
        glue.create_table(
            DatabaseName='sales_analysis_db',
            TableInput={
                'Name': table['name'],
                'StorageDescriptor': {
                    'Columns': [{'Name': col['name'], 'Type': col['type']} for col in table['columns']],
                    'Location': table['path'],
                    'InputFormat': 'org.apache.hadoop.mapred.TextInputFormat',
                    'OutputFormat': 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat',
                    'SerdeInfo': {
                        'SerializationLibrary': 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe',
                        'Parameters': {
                            'field.delim': ',',
                            'serialization.format': ','
                        }
                    }
                },
                'TableType': 'EXTERNAL_TABLE'
            }
        )
        print(f"表 {table['name']} 创建成功")
    except glue.exceptions.AlreadyExistsException:
        print(f"表 {table['name']} 已存在")

# 关闭Spark会话
spark.stop()

print("大数据分析完成，结果已保存到S3并在Glue Catalog中创建了表")