07_dataframe_storage.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "<img src=\"images/dask_horizontal.svg\" align=\"right\" width=\"30%\">"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# 数据存储"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "<img src=\"images/hdd.jpg\" width=\"20%\" align=\"right\">\n",
    "高效的存储可以显着提高性能，尤其是在从磁盘重复操作时。\n",
    "\n",
    "解压文本和解析 CSV 文件的成本很高。 处理中等数据的最有效策略之一是使用二进制存储格式，如 HDF5。 通常这样做的性能提升就足够了，因此您可以再次切换回使用 Pandas 而不是使用 `dask.dataframe`。\n",
    "\n",
    "在本节中，我们将学习如何以磁盘上的二进制格式有效地排列和存储数据集。 我们将使用以下内容：\n",
    "\n",
    "1.  [Pandas `HDFStore`](http://pandas.pydata.org/pandas-docs/stable/io.html#io-hdf5) `HDF5`格式\n",
    "2.  categorical，用于以数值方式存储文本中的数据\n",
    "\n",
    "**要点**\n",
    "\n",
    "1.  存储格式能在数量级上影响性能\n",
    "2.  文本数据即使是使用像HDF5这样的快速格式，保存地也很慢\n",
    "3.  二进制格式、列存储和分区数据的组合将一秒的等待时间变成了 80 毫秒的等待时间。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 创建数据"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "source": [
    "%run prep.py -d accounts"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 读取 CSV"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "首先，我们像以前一样读取我们的 csv 数据。\n",
    "\n",
    "CSV和其他基于文本的文件格式，是存储来自许多来源的数据的最常见的格式，因为它们需要最少的预处理，可以逐行写入并且是人类可读的。 由于 Pandas 的 `read_csv` 得到了很好的优化，CSV 是一个合理，但远非最优的输入，因为读取需要大量的文本解析。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "source": [
    "import os\n",
    "filename = os.path.join('data', 'accounts.*.csv')\n",
    "filename"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "'data/accounts.*.csv'"
      ]
     },
     "metadata": {},
     "execution_count": 2
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "source": [
    "import dask.dataframe as dd\n",
    "df_csv = dd.read_csv(filename)\n",
    "df_csv.head()"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "    id  names  amount\n",
       "0   11    Dan    5549\n",
       "1  128  Sarah    1424\n",
       "2  374    Ray    1069\n",
       "3  374    Ray     687\n",
       "4   11    Dan    5815"
      ],
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>names</th>\n",
       "      <th>amount</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>11</td>\n",
       "      <td>Dan</td>\n",
       "      <td>5549</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>128</td>\n",
       "      <td>Sarah</td>\n",
       "      <td>1424</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>374</td>\n",
       "      <td>Ray</td>\n",
       "      <td>1069</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>374</td>\n",
       "      <td>Ray</td>\n",
       "      <td>687</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>11</td>\n",
       "      <td>Dan</td>\n",
       "      <td>5815</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ]
     },
     "metadata": {},
     "execution_count": 3
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 写入为HDF5"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "HDF5 和 netCDF 是科学领域中非常常用的二进制数组格式。\n",
    "\n",
    "Pandas 包含一个专门的 HDF5 格式，`HDFStore`。 `dd.DataFrame.to_hdf`方法与 `pd.DataFrame.to_hdf` 方法完全一样。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "source": [
    "target = os.path.join('data', 'accounts.h5')\n",
    "target"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "'data/accounts.h5'"
      ]
     },
     "metadata": {},
     "execution_count": 4
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "source": [
    "# 转换为二进制格式，需要花费一些时间\n",
    "%time df_csv.to_hdf(target, '/data')"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 3.67 s, sys: 377 ms, total: 4.05 s\n",
      "Wall time: 6.56 s\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "['data/accounts.h5', 'data/accounts.h5', 'data/accounts.h5']"
      ]
     },
     "metadata": {},
     "execution_count": 5
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "source": [
    "# 和之前一样的数据\n",
    "df_hdf = dd.read_hdf(target, '/data')\n",
    "df_hdf.head()"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "    id  names  amount\n",
       "0   11    Dan    5549\n",
       "1  128  Sarah    1424\n",
       "2  374    Ray    1069\n",
       "3  374    Ray     687\n",
       "4   11    Dan    5815"
      ],
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>id</th>\n",
       "      <th>names</th>\n",
       "      <th>amount</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>11</td>\n",
       "      <td>Dan</td>\n",
       "      <td>5549</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>128</td>\n",
       "      <td>Sarah</td>\n",
       "      <td>1424</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>374</td>\n",
       "      <td>Ray</td>\n",
       "      <td>1069</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>374</td>\n",
       "      <td>Ray</td>\n",
       "      <td>687</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>11</td>\n",
       "      <td>Dan</td>\n",
       "      <td>5815</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ]
     },
     "metadata": {},
     "execution_count": 6
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 比较 CSV 和 HDF5 的速度"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们做了一个简单的计算，需要读取我们数据集的一列，并比较 CSV 文件和我们新创建的 HDF5 文件之间的性能。 你觉得哪个更快？"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "source": [
    "%time df_csv.amount.sum().compute()"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 714 ms, sys: 94.9 ms, total: 809 ms\n",
      "Wall time: 579 ms\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "3005706893"
      ]
     },
     "metadata": {},
     "execution_count": 7
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "source": [
    "%time df_hdf.amount.sum().compute()"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 2.4 s, sys: 298 ms, total: 2.69 s\n",
      "Wall time: 2.8 s\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "3005706893"
      ]
     },
     "metadata": {},
     "execution_count": 8
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "可悲的是，它们大致相同，甚至可能更慢。\n",
    "\n",
    "这里的罪魁祸首是 `names` 列，它是 `object` dtype，因此难以有效存储。 这里有两个问题：\n",
    "\n",
    "1. 我们如何在磁盘上有效地存储像`names` 这样的文本数据？\n",
    "2. 当我们想要的只是`amount` 时，为什么我们必须阅读`names` 列"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 1. 使用categorical有效地存储文本|"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们可以使用 Pandas categoricals 用数字表示替换我们的对象数据类型。 这需要更多的时间，但会带来更好的性能。\n",
    "\n",
    "更多可参考[pandas 文档](http://pandas.pydata.org/pandas-docs/stable/categorical.html) 和 [这篇博文](http://matthewrocklin.com/blog/work/2015/06/18/Categoricals)。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "source": [
    "# 对数据进行分类化(categorize)，然后存储在 HDFStore 中\n",
    "%time df_hdf.categorize(columns=['names']).to_hdf(target, '/data2')"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 6.28 s, sys: 597 ms, total: 6.87 s\n",
      "Wall time: 7.61 s\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "['data/accounts.h5', 'data/accounts.h5', 'data/accounts.h5']"
      ]
     },
     "metadata": {},
     "execution_count": 9
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 看起来是一样的\n",
    "df_hdf = dd.read_hdf(target, '/data2')\n",
    "df_hdf.head()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "source": [
    "# 但加载速度更快\n",
    "%time df_hdf.amount.sum().compute()"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 2.28 s, sys: 168 ms, total: 2.45 s\n",
      "Wall time: 2.76 s\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "3005706893"
      ]
     },
     "metadata": {},
     "execution_count": 10
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "这现在肯定比以前更快。 这告诉我们，不仅是我们使用的文件类型，还有我们如何表示，都会影响存储性能。\n",
    "\n",
    "读取的性能如何取决于我们使用的调度器？ 您可以尝试使用多线程(`threaded`)、多进程(`processes`)和分布式(`distributed`)。\n",
    "\n",
    "然而，这仍然可以更好。 我们必须读取所有列（`names` 和 `amount`）以计算一个（`amount`）的总和。 我们将使用\"parquet\"（一种磁盘列存储格式）进一步改进这一点。 首先，我们了解如何在 dask.dataframe 中设置索引。\n",
    "\n",
    "译者注：\n",
    "如前文所述，传统的csv文件是按行存储和读写的。当数据是按单列而不是多行进行连续存储时，就是所谓的列存储。\n",
    "相对行存储，列存储存在以下优势：\n",
    " - 可以跳过不符合条件的数据，只读取需要的数据，降低 IO 数据量。\n",
    " - 压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码（例如 Run Length Encoding 和 Delta Encoding）进一步节约存储空间。\n",
    " - 只读取需要的列，支持向量运算，能够获取更好的扫描性能。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 练习"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "`fastparquet` 是一个用于与 parquet 格式文件交互的库，它是大数据生态系统中非常常见的格式，被 Hadoop、Spark 和 Impala 等工具使用。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "source": [
    "target = os.path.join('data', 'accounts.parquet')\n",
    "df_csv.categorize(columns=['names']).to_parquet(target, storage_options={\"has_nulls\": True}, engine=\"fastparquet\")"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "调查生成的新目录中的文件结构 - 您认为这些文件是做什么用的？\n",
    "\n",
    "`to_parquet` 带有许多选项，例如压缩、是否显式写入 NULL 信息（在这种情况下不需要）以及如何对字符串进行编码。 您可以在下面试验这些，看看它们对文件大小和处理时间有什么影响，。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "source": [
    "!ls -l data/accounts.parquet/"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "总用量 73256\n",
      "-rw-rw-r-- 1 vmware vmware      952 8月   5 10:39 _common_metadata\n",
      "-rw-rw-r-- 1 vmware vmware     2269 8月   5 10:39 _metadata\n",
      "-rw-rw-r-- 1 vmware vmware 25001753 8月   5 10:39 part.0.parquet\n",
      "-rw-rw-r-- 1 vmware vmware 25001753 8月   5 10:39 part.1.parquet\n",
      "-rw-rw-r-- 1 vmware vmware 25001753 8月   5 10:39 part.2.parquet\n"
     ]
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "source": [
    "df_p = dd.read_parquet(target)\n",
    "# 请注意，列名显示了值的类型 - \n",
    "# 我们可以选择是否加载为分类列。\n",
    "df_p.dtypes"
   ],
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "id           int64\n",
       "names     category\n",
       "amount       int64\n",
       "dtype: object"
      ]
     },
     "metadata": {},
     "execution_count": 14
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "为这个版本的数据重新运行上面的总和计算，并计算需要多长时间。 您可能想多次尝试此操作 - 许多库在第一次调用时执行各种设置工作是很常见的。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "source": [
    "%time df_p.amount.sum().compute()"
   ],
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "CPU times: user 19.2 ms, sys: 24.3 ms, total: 43.5 ms\n",
      "Wall time: 35.1 ms\n"
     ]
    },
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "3005706893"
      ]
     },
     "metadata": {},
     "execution_count": 17
    }
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "归档数据时，通常会按具有唯一标识符的列进行排序和分区，以方便以后快速查找。 对于此数据，该列是`id`。 \n",
    "计算原始 CSV、HDF5 和 parquet 版本以及最终从应用 `set_index('id')` 后编写的新 Parquet 版本检索与 `id==100` 对应的行所需的时间。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "source": [
    "# df_p.set_index('id').to_parquet(...)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 远程文件"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "Dask 可以访问各种面向云和集群的数据存储服务，例如 Amazon S3 或 HDFS\n",
    "\n",
    "优点：\n",
    "* 可扩展、安全的存储\n",
    "\n",
    "缺点：\n",
    "* 网速成为瓶颈"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "设置DataFrame（和其他集合）的方式与以前非常相似。 请注意，这里的数据是匿名可用的，但通常可以传递一个额外的参数`storage_options=`，其中包含有关如何与远程存储交互的更多详细信息。\n",
    "\n",
    "```python\n",
    "taxi = dd.read_csv('s3://nyc-tlc/trip data/yellow_tripdata_2015-*.csv',\n",
    "                   storage_options={'anon': True})\n",
    "```"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "**警告**：互联网上的操作可能需要很长时间才能运行。 此类操作在云集群设置中非常有效，例如从 S3 读取的亚马逊 EC2 机器或从 GCS 读取的 Google 计算机。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [],
   "metadata": {}
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.8.10 64-bit ('dask-tutorial': conda)"
  },
  "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.8.10"
  },
  "interpreter": {
   "hash": "3f2e3fb42896b6b47c37375014a159d976fd49965917325c599b06eb4946867a"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}