{ "cells": [ { "cell_type": "markdown", "id": "bb3b823c-83bd-4bfa-bfbf-2d0f3313bd52", "metadata": {}, "source": [ "# Пример использования Iceberg Metastore с JupyterHub" ] }, { "cell_type": "code", "execution_count": 79, "id": "6b87bc96-51a4-4e9f-a4a1-4f3e59cae246", "metadata": { "scrolled": true }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Package Version\n", "------------------------- --------------\n", "aiobotocore 2.24.2\n", "aiohappyeyeballs 2.6.1\n", "aiohttp 3.12.15\n", "aioitertools 0.12.0\n", "aiosignal 1.4.0\n", "alembic 1.14.0\n", "annotated-types 0.7.0\n", "anyio 4.8.0\n", "argon2-cffi 23.1.0\n", "argon2-cffi-bindings 21.2.0\n", "arrow 1.3.0\n", "asttokens 3.0.0\n", "async-lru 2.0.4\n", "attrs 24.3.0\n", "babel 2.16.0\n", "beautifulsoup4 4.12.3\n", "bleach 6.2.0\n", "boto3 1.40.18\n", "botocore 1.40.18\n", "cachetools 5.5.2\n", "certifi 2024.12.14\n", "certipy 0.2.1\n", "cffi 1.17.1\n", "charset-normalizer 3.4.1\n", "click 8.2.1\n", "comm 0.2.2\n", "cryptography 44.0.0\n", "debugpy 1.8.11\n", "decorator 5.1.1\n", "defusedxml 0.7.1\n", "executing 2.1.0\n", "fastjsonschema 2.21.1\n", "fqdn 1.5.1\n", "frozenlist 1.7.0\n", "fsspec 2025.9.0\n", "greenlet 3.1.1\n", "h11 0.14.0\n", "httpcore 1.0.7\n", "httpx 0.28.1\n", "idna 3.10\n", "ipykernel 6.29.5\n", "ipython 8.31.0\n", "ipython-genutils 0.2.0\n", "isoduration 20.11.0\n", "jedi 0.19.2\n", "Jinja2 3.1.5\n", "jmespath 1.0.1\n", "json5 0.10.0\n", "jsonpointer 3.0.0\n", "jsonschema 4.23.0\n", "jsonschema-specifications 2024.10.1\n", "jupyter_client 8.6.3\n", "jupyter_core 5.7.2\n", "jupyter-events 0.11.0\n", "jupyter-lsp 2.2.5\n", "jupyter_server 2.15.0\n", "jupyter_server_terminals 0.5.3\n", "jupyterhub 5.2.1\n", "jupyterlab 4.3.4\n", "jupyterlab_pygments 0.3.0\n", "jupyterlab_server 2.27.3\n", "Mako 1.3.8\n", "markdown-it-py 4.0.0\n", "MarkupSafe 3.0.2\n", "matplotlib-inline 0.1.7\n", "mdurl 0.1.2\n", "mistune 3.1.0\n", "mmh3 5.2.0\n", "multidict 6.6.4\n", "nbclassic 1.1.0\n", "nbclient 0.10.2\n", "nbconvert 7.16.5\n", "nbformat 5.10.4\n", "nbgitpuller 1.2.1\n", "nest-asyncio 1.6.0\n", "notebook_shim 0.2.4\n", "numpy 2.3.3\n", "oauthlib 3.2.2\n", "overrides 7.7.0\n", "packaging 24.2\n", "pamela 1.2.0\n", "pandas 2.3.2\n", "pandocfilters 1.5.1\n", "parso 0.8.4\n", "pexpect 4.9.0\n", "pip 24.3.1\n", "platformdirs 4.3.6\n", "prometheus_client 0.21.1\n", "prompt_toolkit 3.0.48\n", "propcache 0.3.2\n", "psutil 6.1.1\n", "psycopg2-binary 2.9.10\n", "ptyprocess 0.7.0\n", "pure_eval 0.2.3\n", "pyarrow 21.0.0\n", "pycparser 2.22\n", "pydantic 2.10.5\n", "pydantic_core 2.27.2\n", "Pygments 2.19.1\n", "pyiceberg 0.9.1\n", "pyparsing 3.2.4\n", "python-dateutil 2.9.0.post0\n", "python-json-logger 3.2.1\n", "pytz 2025.2\n", "PyYAML 6.0.2\n", "pyzmq 26.2.0\n", "referencing 0.35.1\n", "requests 2.32.3\n", "rfc3339-validator 0.1.4\n", "rfc3986-validator 0.1.1\n", "rich 13.9.4\n", "rpds-py 0.22.3\n", "s3fs 2025.9.0\n", "s3transfer 0.13.1\n", "Send2Trash 1.8.3\n", "setuptools 75.8.0\n", "six 1.17.0\n", "sniffio 1.3.1\n", "sortedcontainers 2.4.0\n", "soupsieve 2.6\n", "SQLAlchemy 2.0.37\n", "stack-data 0.6.3\n", "strictyaml 1.7.3\n", "tenacity 9.1.2\n", "terminado 0.18.1\n", "tinycss2 1.4.0\n", "tornado 6.4.2\n", "traitlets 5.14.3\n", "types-python-dateutil 2.9.0.20241206\n", "typing_extensions 4.12.2\n", "tzdata 2025.2\n", "uri-template 1.3.0\n", "urllib3 2.3.0\n", "wcwidth 0.2.13\n", "webcolors 24.11.1\n", "webencodings 0.5.1\n", "websocket-client 1.8.0\n", "wrapt 1.17.3\n", "yarl 1.20.1\n", "Note: you may need to restart the kernel to use updated packages.\n" ] } ], "source": [ "pip list " ] }, { "cell_type": "markdown", "id": "1145bf1f-b2bd-4fe9-a4ad-9a07273f8cee", "metadata": {}, "source": [ "### Необходимые переменные" ] }, { "cell_type": "code", "execution_count": null, "id": "74e23270-b207-4693-98a5-bd09ee726c04", "metadata": {}, "outputs": [], "source": [ "uri = \"postgresql://:@/\"\n", "s3_bucket_name = \"\"\n", "db_name = \"\" # которую указывали при создании metastore\n", "ENDPOINT = \"https://hb.bizmrg.com\"\n", "ACCESS_KEY = \"\"\n", "SECRET_KEY = \"\"" ] }, { "cell_type": "markdown", "id": "02a8150f-8819-4ddd-b304-eee0e1e9c19b", "metadata": {}, "source": [ "### Создание namespace в Iceberg" ] }, { "cell_type": "code", "execution_count": 181, "id": "166b26d6-45cd-4cab-afb3-9a13e25fb3de", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Namespace metastore создан успешно!\n" ] } ], "source": [ "from pyiceberg.catalog import load_catalog\n", " \n", "catalog = load_catalog(\n", " \"my_catalog\",\n", " **{\n", " \"type\": \"sql\",\n", " \"uri\": uri,\n", " \"warehouse\": f\"s3://{s3_bucket_name}/datatest123\"\n", " }\n", ")\n", "\n", "try:\n", " catalog.create_namespace(db_name)\n", " print(f\"Namespace {db_name} создан успешно!\")\n", "except Exception as e:\n", " print(f\"Ошибка при создании namespace: {e}\")" ] }, { "cell_type": "markdown", "id": "01d31513-84a8-42b1-8b54-896337d1cdc7", "metadata": {}, "source": [ " ### Вывести все неймспейсы в Iceberg" ] }, { "cell_type": "code", "execution_count": 182, "id": "e7d97d7e-dbb7-4e25-b6e2-2f440f4e7225", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "[('metastore',)]" ] }, "execution_count": 182, "metadata": {}, "output_type": "execute_result" } ], "source": [ "catalog.list_namespaces()" ] }, { "cell_type": "markdown", "id": "fd4ae16e-ad0c-4ec4-aaf0-45ea5b7f3007", "metadata": {}, "source": [ "### Вывести все таблицы" ] }, { "cell_type": "code", "execution_count": 183, "id": "0ef016fa-4a8b-479d-a663-ff8143e69718", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "[]" ] }, "execution_count": 183, "metadata": {}, "output_type": "execute_result" } ], "source": [ "catalog.list_tables(db_name)" ] }, { "cell_type": "markdown", "id": "e0aeaff2-7dc4-4d1e-a39b-5c389e51ef90", "metadata": {}, "source": [ "### Проверка подключения к S3" ] }, { "cell_type": "code", "execution_count": 184, "id": "e2117ca7-00e4-45bd-8152-6fb3eef39397", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Подключение к S3 успешно. Bucket: jh-test-iceberg\n", "Bucket не пустой\n", "Первые объекты:\n", " - datatest123/\n" ] } ], "source": [ "import boto3\n", "\n", "try:\n", " s3_client = boto3.client(\n", " 's3',\n", " endpoint_url=ENDPOINT,\n", " aws_access_key_id=ACCESS_KEY,\n", " aws_secret_access_key=SECRET_KEY\n", " )\n", " \n", " response = s3_client.list_objects_v2(Bucket=s3_bucket_name, MaxKeys=1)\n", " print(f\"Подключение к S3 успешно. Bucket: {s3_bucket_name}\")\n", " \n", " if 'Contents' in response:\n", " print(\"Bucket не пустой\")\n", " print(\"Первые объекты:\")\n", " for obj in response['Contents'][:3]:\n", " print(f\" - {obj['Key']}\")\n", " else:\n", " print(\"Bucket пустой\")\n", " \n", "except Exception as e:\n", " print(f\"Ошибка подключения к S3: {e}\")" ] }, { "cell_type": "code", "execution_count": 174, "id": "3243dde3-e5b4-40c0-b668-fef0035ec03d", "metadata": {}, "outputs": [], "source": [ "import os\n", "import boto3\n", "import pyarrow as pa\n", "from pyiceberg.catalog import load_catalog\n", "from pyiceberg.schema import Schema\n", "from pyiceberg.types import NestedField, StringType, IntegerType\n", "import pandas as pd\n", "from typing import Union" ] }, { "cell_type": "markdown", "id": "252bc234-3342-42b7-af67-a9f5e56e664f", "metadata": {}, "source": [ "### Функция по приведению типов данных pandas DataFrame для совместимости с Iceberg" ] }, { "cell_type": "code", "execution_count": 150, "id": "6c9392f6-7735-48f3-b840-2a93f5fbbfd5", "metadata": {}, "outputs": [], "source": [ "def prepare_dataframe_for_iceberg(data_df: pd.DataFrame) -> pd.DataFrame:\n", " \"\"\"\n", " Приводит типы данных pandas DataFrame для совместимости с Iceberg\n", " \n", " Args:\n", " data_df: Исходный pandas DataFrame\n", " \n", " Returns:\n", " DataFrame с приведенными типами данных\n", " \"\"\"\n", " prepared_df = data_df.copy()\n", " \n", " for col in prepared_df.columns:\n", " dtype = prepared_df[col].dtype\n", " \n", " if dtype == 'object':\n", " prepared_df[col] = prepared_df[col].astype('string')\n", " \n", " elif pd.api.types.is_integer_dtype(dtype):\n", " # Конвертируем все int типы в int64 для совместимости с Iceberg\n", " prepared_df[col] = prepared_df[col].astype('int64')\n", " \n", " elif pd.api.types.is_float_dtype(dtype):\n", " prepared_df[col] = prepared_df[col].astype('float64')\n", " \n", " \n", " elif pd.api.types.is_bool_dtype(dtype):\n", " prepared_df[col] = prepared_df[col].astype('boolean')\n", " \n", " elif pd.api.types.is_datetime64_any_dtype(dtype):\n", " prepared_df[col] = prepared_df[col].astype('datetime64[us]')\n", " \n", " return prepared_df\n", " " ] }, { "cell_type": "markdown", "id": "8607aa47-f94e-408f-9a37-99e566ec9e02", "metadata": {}, "source": [ "### Функция для создания PyArrow схемы" ] }, { "cell_type": "code", "execution_count": 151, "id": "06843416-18eb-4131-91b8-ca0665d059a9", "metadata": {}, "outputs": [], "source": [ "def create_pyarrow_scheme(prepared_df: pd.DataFrame) -> pa.schema:\n", " fields = []\n", " for col in prepared_df.columns:\n", " if pd.api.types.is_datetime64_any_dtype(prepared_df[col].dtype):\n", " fields.append(pa.field(col, pa.timestamp('us')))\n", " elif prepared_df[col].dtype == 'string':\n", " fields.append(pa.field(col, pa.string()))\n", " elif prepared_df[col].dtype == 'int64':\n", " fields.append(pa.field(col, pa.int64()))\n", " elif prepared_df[col].dtype == 'float64':\n", " fields.append(pa.field(col, pa.float64()))\n", " elif prepared_df[col].dtype == 'boolean':\n", " fields.append(pa.field(col, pa.bool_()))\n", " else:\n", " fields.append(pa.field(col, pa.from_pandas_dtype(prepared_df[col].dtype)))\n", " \n", " return pa.schema(fields)" ] }, { "cell_type": "markdown", "id": "0627b1a5-0738-4044-8127-2e27769296cc", "metadata": {}, "source": [ "### Функция добавления pandas DataFrame в Iceberg таблицу" ] }, { "cell_type": "code", "execution_count": 153, "id": "f9e444b9-030d-4e9a-ab44-a17d77298687", "metadata": {}, "outputs": [], "source": [ "from pyiceberg.table import Table\n", "import pandas as pd\n", "import pyarrow as pa\n", "\n", "def add_data_to_iceberg_table(table: Table, data_df: pd.DataFrame) -> None:\n", " \"\"\"\n", " Добавляет pandas DataFrame в Iceberg таблицу\n", "\n", " Args:\n", " table: Iceberg таблица\n", " data_df: pandas DataFrame для добавления\n", " \"\"\"\n", " # Подготавливаем данные для Iceberg\n", " prepared_df = prepare_dataframe_for_iceberg(data_df)\n", " \n", " # Создаем PyArrow схему\n", " schema = create_pyarrow_scheme(prepared_df)\n", " \n", " arrow_table = pa.Table.from_pandas(prepared_df, schema=schema)\n", "\n", " table.append(arrow_table)\n", "\n", " print(f\"Добавлено {len(data_df)} записей в таблицу\")\n", " " ] }, { "cell_type": "markdown", "id": "a719b777-100b-4688-9226-b1fe991cd4bb", "metadata": {}, "source": [ "### Создание каталога" ] }, { "cell_type": "markdown", "id": "6d5eb4c7-a550-4ee9-87c9-efc4a9fb1c50", "metadata": {}, "source": [ "Можем указать свою папку в бакете в строке warehouse=f\"s3://{s3_bucket_name}/datatest123/test_table\"" ] }, { "cell_type": "code", "execution_count": null, "id": "3a2de5d7-2dcb-4cab-a00d-de4671222b33", "metadata": {}, "outputs": [], "source": [ "# Создаем каталог\n", "catalog = load_catalog(\n", " \"my_catalog\",\n", " **{\n", " \"type\": \"sql\",\n", " \"uri\": uri,\n", " \"warehouse\": f\"s3://{s3_bucket_name}/\",\n", " \"s3.endpoint\": ENDPOINT,\n", " \"s3.access-key-id\": ACCESS_KEY,\n", " \"s3.secret-access-key\": SECRET_KEY\n", " }\n", ")" ] }, { "cell_type": "markdown", "id": "ee0fc7a1-3611-46d7-a83f-696ad0895ece", "metadata": {}, "source": [ "### Удаляем существующую таблицу" ] }, { "cell_type": "code", "execution_count": 164, "id": "9b52a7ae-9370-44d0-b3ea-6ad9784116e6", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Старая таблица удалена\n" ] } ], "source": [ "# Удаляем существующую таблицу\n", "try:\n", " catalog.drop_table(\"metastore.test_table\")\n", " print(\"Старая таблица удалена\")\n", "except:\n", " print(\"Старая таблица не найдена\")" ] }, { "cell_type": "markdown", "id": "c7532da7-ae2d-420c-a2ba-278b3fba8e5a", "metadata": {}, "source": [ "### Создаем тестовые данные" ] }, { "cell_type": "code", "execution_count": 176, "id": "3626ca70-5398-4a85-8ed6-a8e6b70b81fe", "metadata": {}, "outputs": [], "source": [ "test_data = pd.DataFrame({\n", " 'id': [1, 2, 3], \n", " 'name': ['Alice', 'Bob', 'Charlie'],\n", " 'price': [10.5, 20.3, 30.1],\n", " 'is_active': [True, False, True],\n", " 'created_at': pd.to_datetime(['2024-01-01', '2024-01-02', '2024-01-03'])\n", "})" ] }, { "cell_type": "markdown", "id": "669d58f5-d364-481f-a171-f3769be1ffd1", "metadata": {}, "source": [ "### Определяем схему таблицы" ] }, { "cell_type": "code", "execution_count": 186, "id": "e7ee9153-1042-491c-9c55-c993dd436201", "metadata": {}, "outputs": [], "source": [ "from pyiceberg.types import NestedField, StringType, LongType, DoubleType, BooleanType, TimestampType\n", "\n", "\n", "schema = Schema(\n", " NestedField(1, \"id\", LongType(), required=False),\n", " NestedField(2, \"name\", StringType(), required=False),\n", " NestedField(3, \"price\", DoubleType(), required=False),\n", " NestedField(4, \"is_active\", BooleanType(), required=False),\n", " NestedField(5, \"created_at\", TimestampType(), required=False)\n", ")" ] }, { "cell_type": "markdown", "id": "14171626-2985-43f4-8d1e-a531e6fc5ccf", "metadata": {}, "source": [ "### Создаем таблицу" ] }, { "cell_type": "markdown", "id": "c17574f9-47a1-49c2-a96e-48083e4055da", "metadata": {}, "source": [ "Можем указать свою папку в бакете в строке location=f\"s3://{s3_bucket_name}/datatest123/test_table\"" ] }, { "cell_type": "code", "execution_count": 187, "id": "a4de08b4-f2d1-42e8-8d2b-4dee96972f7f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Таблица 'metastore.test_table' создана успешно!\n" ] } ], "source": [ "try:\n", " table = catalog.create_table(\n", " \"metastore.test_table\",\n", " schema=schema,\n", " location=f\"s3://{s3_bucket_name}/metastore/test_table\"\n", " )\n", " print(\"Таблица 'metastore.test_table' создана успешно!\")\n", " \n", "except Exception as e:\n", " print(f\"Ошибка при создании таблицы: {e}\")" ] }, { "cell_type": "markdown", "id": "339d4c19-1455-4b1a-92f7-7c1575521c2a", "metadata": {}, "source": [ "### Добавляем данные" ] }, { "cell_type": "code", "execution_count": 159, "id": "142e67fc-8291-46d2-a742-9325f697ae3d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Добавлено 3 записей в таблицу\n" ] } ], "source": [ "try:\n", " add_data_to_iceberg_table(table, test_data)\n", "except Exception as e:\n", " print(f\"Ошибка при создании таблицы: {e}\")" ] }, { "cell_type": "markdown", "id": "8ab2840e-baa6-4f35-84dd-0330e8b2b03a", "metadata": {}, "source": [ "### Читаем все данные из таблицы" ] }, { "cell_type": "code", "execution_count": 160, "id": "bfa5b767-cb29-4755-905e-9ecc53000c8c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Все данные в таблице:\n", " id name price is_active created_at\n", "0 1 Alice 10.5 True 2024-01-01\n", "1 2 Bob 20.3 False 2024-01-02\n", "2 3 Charlie 30.1 True 2024-01-03\n" ] } ], "source": [ "df_all = table.scan().to_pandas()\n", "print(\"Все данные в таблице:\")\n", "print(df_all)" ] }, { "cell_type": "markdown", "id": "637836cc-40a7-424f-8352-6e319d6195ff", "metadata": {}, "source": [ "### Добавляем новые данные" ] }, { "cell_type": "code", "execution_count": 161, "id": "e3a013e4-8ba0-40a6-aa6a-f002c393d4cc", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Добавлено 3 записей в таблицу\n" ] } ], "source": [ "new_data = pd.DataFrame({\n", " 'id': [4, 5, 6],\n", " 'name': ['Diana', 'Eve', 'Frank'],\n", " 'price': [40.5, 50.3, 60.1],\n", " 'is_active': [False, True, False],\n", " 'created_at': pd.to_datetime(['2024-01-04', '2024-01-05', '2024-01-06'])\n", "})\n", " \n", "add_data_to_iceberg_table(table, new_data)" ] }, { "cell_type": "markdown", "id": "f9185ded-e37a-4f31-8eb7-6ebd9afcfdf6", "metadata": {}, "source": [ "### Читаем все данные из таблицы" ] }, { "cell_type": "code", "execution_count": 162, "id": "425eb24d-3c6e-43b6-b5e9-8d9e3f75b02a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Все данные в таблице:\n", " id name price is_active created_at\n", "0 4 Diana 40.5 False 2024-01-04\n", "1 5 Eve 50.3 True 2024-01-05\n", "2 6 Frank 60.1 False 2024-01-06\n", "3 1 Alice 10.5 True 2024-01-01\n", "4 2 Bob 20.3 False 2024-01-02\n", "5 3 Charlie 30.1 True 2024-01-03\n" ] } ], "source": [ "df_all = table.scan().to_pandas()\n", "print(\"Все данные в таблице:\")\n", "print(df_all)" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "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.12.8" } }, "nbformat": 4, "nbformat_minor": 5 }