Draft notebook for workflow with vineyard example

Signed-off-by: trafalgarzzz <[email protected]>

Author: TrafalgarZZZ

examples/03_dataflow_with_vineyard/static/workflow.png

@@ -0,0 +1,389 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# 在数据处理流水线中使用VineyardRuntime实现高效中间数据管理"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 概述\n",
+    "当下的大数据/AI应用，往往需要使用端到端的流水线来实现，以下图所示的一个风控作业数据操作流为例：1首先，需要从数据库中导出订单相关数据；随后，图计算引擎会处理这些原始数据，构建 用户-商品 关系图，并通过图算法，初筛出其中隐藏的潜在作弊团伙；接下来，机器学习算法会对这些潜在团伙进行作弊归因，筛选出更准确的结果；最后这些结果会经过人工筛查，并最终做出业务处理。\n",
+    "\n",
+    "![Workflow](./static/workflow.png)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "在这样的场景下，我们常常会遇到如下问题：\n",
+    "1. 开发环境和生产环境的差异导致数据工作流的开发和调试变得复杂且低效：\n",
+    "数据科学家在自己的计算机上开发数据操作的操作使用 Python 代码，但是又需要在生产环境中将代码转化为他们并不熟悉的 YAML 文件从而利用 Argo、Tekton 等基于 Kubernetes 的工作流引擎，这大大降低了开发和部署效率，也带来了开发和生产环境差异性大带来的风险。\n",
+    "2. 需要引入新分布式存储实现中间临时数据交换，带来额外的开发、费用、运维成本：\n",
+    "端到端任务的子任务之间的数据交换通常依赖分布式文件系统或对象存储系统（如 HDFS、S3、OSS），这使得整个工作流需要进行大量的数据格式转换和适配工作，导致冗余的 I/O 操作，并由于中间数据的短期性，使用分布式存储系统会导致额外的成本。\n",
+    "\n",
+    "3. 在大规模 Kubernetes 集群环境中的数据处理的效率问题：\n",
+    "在大规模的 Kubernetes 集群中，使用现有的分布式文件系统处理数据时，由于调度系统对数据的读写本地性缺乏足够的理解，并未有效地考虑到数据的位置问题，没有充分利用数据的局部性，导致在处理节点间的数据交换时，无法避免大量的数据重复拉取操作。这种操作既增加了 I/O 消耗，也降低了整体的运行效率。"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "![workflow with vineyard](./static/workflow_with_vineyard.png)\n",
+    "为了解决现有大数据/AI中的数据流操作存在的上述问题，我们结合了 Vineyard 的数据共享机制和 Fluid的数据编排能力。\n",
+    "1. Fluid 的 Python SDK 能够方便地对数据流进行编排，为熟悉 Python 的数据科学家提供了一种简单的方式来构建和提交以数据集操作为中心的工作流。特别地，在开发环境和云上生产环境通过一套代码进行数据流管理。\n",
+    "2. Vineyard 使端到端工作流中任务之间的数据共享更加高效， 通过内存映射的方式实现零拷贝数据共享，从而避免了额外的 IO 开销，这个是数据共享效率提升的关键。\n",
+    "3. 通过利用 Fluid 的数据亲和性调度能力，在 Pod 调度策略考虑数据写入节点的信息，从而减小数据迁移引入的网络开销，提升端到端性能。"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 代码示例\n",
+    "\n",
+    "在接下来的示例中，我们将使用Fluid中的VineyardRuntime以及DataFlow功能展示如何在数据处理流水线中实现高效中间数据管理。DataFlow功能是Fluid内建提供的数据流编排能力，可将数据处理过程中的多个数据操作串联，实现简单的逻辑编排。如果希望使用更为高级的工作流编排能力，VineyardRuntime同样支持与Argo Workflow等工作流编排引擎集成使用。"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 1. 数据集准备"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import numpy as np\n",
+    "import pandas as pd\n",
+    "\n",
+    "# 生成大小约为22G的dataframe\n",
+    "num_rows = 6000 * 10000\n",
+    "df = pd.DataFrame({\n",
+    "    'Id': np.random.randint(1, 100000, num_rows),\n",
+    "    'MSSubClass': np.random.randint(20, 201, size=num_rows),\n",
+    "    'LotFrontage': np.random.randint(50, 151, size=num_rows),\n",
+    "    'LotArea': np.random.randint(5000, 20001, size=num_rows),\n",
+    "    'OverallQual': np.random.randint(1, 11, size=num_rows),\n",
+    "    'OverallCond': np.random.randint(1, 11, size=num_rows),\n",
+    "    'YearBuilt': np.random.randint(1900, 2022, size=num_rows),\n",
+    "    'YearRemodAdd': np.random.randint(1900, 2022, size=num_rows),\n",
+    "    'MasVnrArea': np.random.randint(0, 1001, size=num_rows),\n",
+    "    'BsmtFinSF1': np.random.randint(0, 2001, size=num_rows),\n",
+    "    'BsmtFinSF2': np.random.randint(0, 1001, size=num_rows),\n",
+    "    'BsmtUnfSF': np.random.randint(0, 2001, size=num_rows),\n",
+    "    'TotalBsmtSF': np.random.randint(0, 3001, size=num_rows),\n",
+    "    '1stFlrSF': np.random.randint(500, 4001, size=num_rows),\n",
+    "    '2ndFlrSF': np.random.randint(0, 2001, size=num_rows),\n",
+    "    'LowQualFinSF': np.random.randint(0, 201, size=num_rows),\n",
+    "    'GrLivArea': np.random.randint(600, 5001, size=num_rows),\n",
+    "    'BsmtFullBath': np.random.randint(0, 4, size=num_rows),\n",
+    "    'BsmtHalfBath': np.random.randint(0, 3, size=num_rows),\n",
+    "    'FullBath': np.random.randint(0, 5, size=num_rows),\n",
+    "    'HalfBath': np.random.randint(0, 3, size=num_rows),\n",
+    "    'BedroomAbvGr': np.random.randint(0, 11, size=num_rows),\n",
+    "    'KitchenAbvGr': np.random.randint(0, 4, size=num_rows),\n",
+    "    'TotRmsAbvGrd': np.random.randint(0, 16, size=num_rows),\n",
+    "    'Fireplaces': np.random.randint(0, 4, size=num_rows),\n",
+    "    'GarageYrBlt': np.random.randint(1900, 2022, size=num_rows),\n",
+    "    'GarageCars': np.random.randint(0, 5, num_rows),\n",
+    "    'GarageArea': np.random.randint(0, 1001, num_rows),\n",
+    "    'WoodDeckSF': np.random.randint(0, 501, num_rows),\n",
+    "    'OpenPorchSF': np.random.randint(0, 301, num_rows),\n",
+    "    'EnclosedPorch': np.random.randint(0, 201, num_rows),\n",
+    "    '3SsnPorch': np.random.randint(0, 101, num_rows),\n",
+    "    'ScreenPorch': np.random.randint(0, 201, num_rows),\n",
+    "    'PoolArea': np.random.randint(0, 301, num_rows),\n",
+    "    'MiscVal': np.random.randint(0, 5001, num_rows),\n",
+    "    'TotalRooms': np.random.randint(2, 11, num_rows),\n",
+    "    \"GarageAge\": np.random.randint(1, 31, num_rows),\n",
+    "    \"RemodAge\": np.random.randint(1, 31, num_rows),\n",
+    "    \"HouseAge\": np.random.randint(1, 31, num_rows),\n",
+    "    \"TotalBath\": np.random.randint(1, 5, num_rows),\n",
+    "    \"TotalPorchSF\": np.random.randint(1, 1001, num_rows),\n",
+    "    \"TotalSF\": np.random.randint(1000, 6001, num_rows),\n",
+    "    \"TotalArea\": np.random.randint(1000, 6001, num_rows),\n",
+    "    'MoSold': np.random.randint(1, 13, num_rows),\n",
+    "    'YrSold': np.random.randint(2006, 2022, num_rows),\n",
+    "    'SalePrice': np.random.randint(50000, 800001, num_rows),\n",
+    "})\n",
+    "\n",
+    "import oss2\n",
+    "import io\n",
+    "from oss2.credentials import EnvironmentVariableCredentialsProvider\n",
+    "# 请将您的 OSS accessKeyID 和 accessKeySecret 分别设置成环境变量 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET\n",
+    "auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())\n",
+    "# 请将 OSS_ENDPOINT 和 BUCKET_NAME 替换为您的 OSS Endpoint 和 Bucket\n",
+    "bucket = oss2.Bucket(auth, 'OSS_ENDPOINT', 'BUCKET_NAME')\n",
+    "\n",
+    "bytes_buffer = io.BytesIO()\n",
+    "df.to_pickle(bytes_buffer)\n",
+    "bucket.put_object(\"df.pkl\", bytes_buffer.getvalue())"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 2. 创建Fluid Dataset和VineyardRuntime"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import fluid\n",
+    "\n",
+    "from fluid import constants\n",
+    "from fluid import models\n",
+    "\n",
+    "# 使用默认kubeconfig文件连接到 Fluid 控制平台，并创建 Fluid 客户端实例\n",
+    "client_config = fluid.ClientConfig()\n",
+    "fluid_client = fluid.FluidClient(client_config)\n",
+    "\n",
+    "# 在default namespace下创建名为vineyard的数据集\n",
+    "fluid_client.create_dataset(\n",
+    "    dataset_name=\"vineyard\",\n",
+    ")\n",
+    "\n",
+    "# 获取vineyard数据集实例\n",
+    "dataset = fluid_client.get_dataset(dataset_name=\"vineyard\")\n",
+    "\n",
+    "# 初始化vineyard runtime的配置，并将vineyard数据集实例绑定到该runtime。\n",
+    "# 副本数为2，内存分别为30Gi\n",
+    "dataset.bind_runtime(\n",
+    "    runtime_type=constants.VINEYARD_RUNTIME_KIND,\n",
+    "    replicas=2,\n",
+    "    cache_capacity_GiB=30,\n",
+    "    cache_medium=\"MEM\",\n",
+    "    wait=True\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "在上述代码片段中：\n",
+    "- 创建 Fluid 客户端: 这段代码负责使用默认的kubeconfig文件建立与Fluid控制平台的连接，并创建一个Fluid客户端实例。\n",
+    "- 创建和配置 vineyard 数据集与运行时环境: 接下来，代码创建了一个名为Vineyard的数据集，然后获取该数据集实例，并初始化vineyard运行时的配置，设置副本数和内存大小，将数据集绑定到运行时环境。"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 3. 定义Fluid DataFlow"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from kubernetes.client import models as k8s_models\n",
+    "# 定义任务运行模版，并挂载OSS Volume\n",
+    "def create_processor(script):\n",
+    "    return models.Processor(\n",
+    "        # 当按照前面的可选步骤开启fuse亲和性调度后, 添加下列标签, 从而实现数据处理的最佳性能\n",
+    "        # pod_metadata=models.PodMetadata(\n",
+    "        #     labels={\"fuse.serverful.fluid.io/inject\": \"true\"},\n",
+    "        # ),\n",
+    "        script=models.ScriptProcessor(\n",
+    "            command=[\"bash\"],\n",
+    "            source=script,\n",
+    "            image=\"python\",\n",
+    "            image_tag=\"3.10\",\n",
+    "            volumes=[k8s_models.V1Volume(\n",
+    "                name=\"data\",\n",
+    "                persistent_volume_claim=k8s_models.V1PersistentVolumeClaimVolumeSource(\n",
+    "                    claim_name=\"pvc-oss\"\n",
+    "                )\n",
+    "            )],\n",
+    "            volume_mounts=[k8s_models.V1VolumeMount(\n",
+    "                name=\"data\",\n",
+    "                mount_path=\"/data\"\n",
+    "            )],\n",
+    "        )   \n",
+    "    )"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "在上述代码片段中：\n",
+    "- **创建任务模版:** 代码中封装了一个名为`create_processor`的任务模板函数，该函数接收一个bash脚本并把它传入作为某个容器的启动命令。该容器中定义了Python 3.10的运行环境，并在`/data`目录下挂载了OSS存储数据源。"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 定义数据预处理脚本\n",
+    "preprocess_data_script = \"\"\"\n",
+    "pip3 install numpy pandas pyarrow requests vineyard scikit-learn==1.4.0 joblib==1.3.2\n",
+    "#!/bin/bash\n",
+    "set -ex\n",
+    "\n",
+    "cat <<EOF > ./preprocess.py\n",
+    "from sklearn.model_selection import train_test_split\n",
+    "\n",
+    "import pandas as pd\n",
+    "import vineyard\n",
+    "\n",
+    "df = pd.read_pickle('/data/df.pkl')\n",
+    "\n",
+    "# Preprocess Data\n",
+    "df = df.drop(df[(df['GrLivArea']>4800)].index)\n",
+    "X = df.drop('SalePrice', axis=1)  # Features\n",
+    "y = df['SalePrice']  # Target variable\n",
+    "\n",
+    "del df\n",
+    "\n",
+    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)\n",
+    "\n",
+    "del X, y\n",
+    "\n",
+    "vineyard.put(X_train, name=\"x_train\", persist=True)\n",
+    "vineyard.put(X_test, name=\"x_test\", persist=True)\n",
+    "vineyard.put(y_train, name=\"y_train\", persist=True)\n",
+    "vineyard.put(y_test, name=\"y_test\", persist=True)\n",
+    "\n",
+    "EOF\n",
+    "\n",
+    "python3 ./preprocess.py\n",
+    "\"\"\"\n",
+    "\n",
+    "# 定义模型训练脚本\n",
+    "train_data_script = \"\"\"\n",
+    "pip3 install numpy pandas pyarrow requests vineyard scikit-learn==1.4.0 joblib==1.3.2\n",
+    "#!/bin/bash\n",
+    "set -ex\n",
+    "\n",
+    "cat <<EOF > ./train.py\n",
+    "from sklearn.linear_model import LinearRegression\n",
+    "\n",
+    "import joblib\n",
+    "import pandas as pd\n",
+    "import vineyard\n",
+    "\n",
+    "x_train_data = vineyard.get(name=\"x_train\", fetch=True)\n",
+    "y_train_data = vineyard.get(name=\"y_train\", fetch=True)\n",
+    "\n",
+    "model = LinearRegression()\n",
+    "model.fit(x_train_data, y_train_data)\n",
+    "\n",
+    "joblib.dump(model, '/data/model.pkl')\n",
+    "\n",
+    "EOF\n",
+    "python3 ./train.py\n",
+    "\"\"\"\n",
+    "\n",
+    "# 定义模型测试脚本\n",
+    "test_data_script = \"\"\"\n",
+    "pip3 install numpy pandas pyarrow requests vineyard scikit-learn==1.4.0 joblib==1.3.2\n",
+    "#!/bin/bash\n",
+    "set -ex\n",
+    "\n",
+    "cat <<EOF > ./test.py\n",
+    "from sklearn.linear_model import LinearRegression\n",
+    "from sklearn.metrics import mean_squared_error\n",
+    "\n",
+    "import vineyard\n",
+    "import joblib\n",
+    "import pandas as pd\n",
+    "\n",
+    "x_test_data = vineyard.get(name=\"x_test\", fetch=True)\n",
+    "y_test_data = vineyard.get(name=\"y_test\", fetch=True)\n",
+    "\n",
+    "model = joblib.load(\"/data/model.pkl\")\n",
+    "y_pred = model.predict(x_test_data)\n",
+    "\n",
+    "err = mean_squared_error(y_test_data, y_pred)\n",
+    "\n",
+    "with open('/data/output.txt', 'a') as f:\n",
+    "    f.write(str(err))\n",
+    "\n",
+    "EOF\n",
+    "\n",
+    "python3 ./test.py\n",
+    "\"\"\"\n",
+    "\n",
+    "preprocess_processor = create_processor(preprocess_data_script)\n",
+    "train_processor = create_processor(train_data_script)\n",
+    "test_processor = create_processor(test_data_script)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "上述代码片段分别定义了数据处理流水线中的三个步骤：数据预处理、模型训练和模型测试。这三个步骤对应的Bash脚本传入`create_processor`函数以被封装为三个processor。"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 创建线性回归模型的任务工作流：数据预处理 -> 模型训练 -> 模型测试\n",
+    "# 下列的挂载路径\"/var/run\"是vineyard配置文件的默认路径\n",
+    "flow = dataset.process(processor=preprocess_processor, dataset_mountpath=\"/var/run\") \\\n",
+    "              .process(processor=train_processor, dataset_mountpath=\"/var/run\") \\\n",
+    "              .process(processor=test_processor, dataset_mountpath=\"/var/run\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 将线性回归模型的数据处理任务工作流提交，并等待其运行完成\n",
+    "run = flow.run(run_id=\"linear-regression-with-vineyard\")\n",
+    "run.wait()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 4. 资源清理"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 清理所有资源\n",
+    "dataset.clean_up(wait=True)"
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}