dask-bag.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask Bag\n",
    "\n",
    "Dask-bag 擅长处理可以表示为任意输入序列的数据。我们将其称为“杂乱”数据，因为它可能包含复杂的嵌套结构、缺失的字段、数据类型的混合等。\n",
    "\n",
    "默认情况下，dask.bag使用dask.multiprocessing的计算。作为一个好处，Dask 绕过GIL并在纯 Python 对象上使用多个内核。作为一个缺点，Dask Bag 在包含大量工作间通信的计算中表现不佳。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client, progress\n",
    "client = Client(n_workers=4, threads_per_worker=1)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 创建随机数据\n",
    "\n",
    "创建一组随机的记录数据，并将其以JSON形式存储到磁盘。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['/Users/kefei/demo/dask-learn/data/0.json',\n",
       " '/Users/kefei/demo/dask-learn/data/1.json',\n",
       " '/Users/kefei/demo/dask-learn/data/2.json',\n",
       " '/Users/kefei/demo/dask-learn/data/3.json',\n",
       " '/Users/kefei/demo/dask-learn/data/4.json',\n",
       " '/Users/kefei/demo/dask-learn/data/5.json',\n",
       " '/Users/kefei/demo/dask-learn/data/6.json',\n",
       " '/Users/kefei/demo/dask-learn/data/7.json',\n",
       " '/Users/kefei/demo/dask-learn/data/8.json',\n",
       " '/Users/kefei/demo/dask-learn/data/9.json']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import dask\n",
    "import json\n",
    "import os\n",
    "\n",
    "os.makedirs('data', exist_ok=True)              \n",
    "\n",
    "b = dask.datasets.make_people()                \n",
    "b.map(json.dumps).to_textfiles('data/*.json')  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 读取JSON数据"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"age\": 51, \"name\": [\"Chas\", \"Ratliff\"], \"occupation\": \"Seaman\", \"telephone\": \"344-219-9130\", \"address\": {\"address\": \"1135 Hawthorne Concession road\", \"city\": \"Springfield\"}, \"credit-card\": {\"number\": \"3416 044674 28287\", \"expiration-date\": \"06/20\"}}\n",
      "{\"age\": 53, \"name\": [\"Wilbur\", \"Cannon\"], \"occupation\": \"Geophysicist\", \"telephone\": \"(219) 049-8427\", \"address\": {\"address\": \"871 Bernal Heights Lane\", \"city\": \"Hallandale Beach\"}, \"credit-card\": {\"number\": \"4315 9929 2231 3721\", \"expiration-date\": \"04/22\"}}\n"
     ]
    }
   ],
   "source": [
    "!head -n 2 data/0.json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dask.bag<loads, npartitions=20>"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import dask.bag as db\n",
    "import json\n",
    "\n",
    "b = db.read_text('data/*.json').map(json.loads)\n",
    "b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'age': 51,\n",
       "  'name': ['Chas', 'Ratliff'],\n",
       "  'occupation': 'Seaman',\n",
       "  'telephone': '344-219-9130',\n",
       "  'address': {'address': '1135 Hawthorne Concession road',\n",
       "   'city': 'Springfield'},\n",
       "  'credit-card': {'number': '3416 044674 28287', 'expiration-date': '06/20'}},\n",
       " {'age': 53,\n",
       "  'name': ['Wilbur', 'Cannon'],\n",
       "  'occupation': 'Geophysicist',\n",
       "  'telephone': '(219) 049-8427',\n",
       "  'address': {'address': '871 Bernal Heights Lane',\n",
       "   'city': 'Hallandale Beach'},\n",
       "  'credit-card': {'number': '4315 9929 2231 3721',\n",
       "   'expiration-date': '04/22'}})"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 查看前两个数据\n",
    "b.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map, Filter, Aggregate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'age': 51,\n",
       "  'name': ['Chas', 'Ratliff'],\n",
       "  'occupation': 'Seaman',\n",
       "  'telephone': '344-219-9130',\n",
       "  'address': {'address': '1135 Hawthorne Concession road',\n",
       "   'city': 'Springfield'},\n",
       "  'credit-card': {'number': '3416 044674 28287', 'expiration-date': '06/20'}},\n",
       " {'age': 53,\n",
       "  'name': ['Wilbur', 'Cannon'],\n",
       "  'occupation': 'Geophysicist',\n",
       "  'telephone': '(219) 049-8427',\n",
       "  'address': {'address': '871 Bernal Heights Lane',\n",
       "   'city': 'Hallandale Beach'},\n",
       "  'credit-card': {'number': '4315 9929 2231 3721',\n",
       "   'expiration-date': '04/22'}})"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 查看年龄超过30的人\n",
    "b.filter(lambda record: record['age'] > 30).take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('Seaman', 'Geophysicist')"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 查看每个记录中的occupation值\n",
    "b.map(lambda record: record['occupation']).take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "17025"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 统计数量\n",
    "b.count().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 链式计算"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dask.bag<topk-aggregate, npartitions=1>"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 计算年龄大于30的按occupation排序的前10位\n",
    "result = (b.filter(lambda record: record['age'] > 30)\n",
    "           .map(lambda record: record['occupation'])\n",
    "           .frequencies(sort=True)\n",
    "           .topk(10, key=1))\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Stonemason', 25),\n",
       " ('Nursing Manager', 24),\n",
       " ('Theatrical Agent', 23),\n",
       " ('Horticulturalist', 23),\n",
       " ('Valve Technician', 23),\n",
       " ('Landworker', 22),\n",
       " ('Mortician', 22),\n",
       " ('Blind Fitter', 21),\n",
       " ('Care Assistant', 21),\n",
       " ('Seamstress', 21)]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据转换+存储"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['/Users/kefei/demo/dask-learn/data/processed.00.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.01.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.02.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.03.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.04.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.05.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.06.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.07.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.08.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.09.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.10.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.11.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.12.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.13.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.14.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.15.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.16.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.17.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.18.json',\n",
       " '/Users/kefei/demo/dask-learn/data/processed.19.json']"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(b.filter(lambda record: record['age'] > 30) \n",
    "  .map(json.dumps)                           \n",
    "  .to_textfiles('data/processed.*.json'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 预处理后转换为DataFramesb.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'age': 51,\n",
       "  'name': ['Chas', 'Ratliff'],\n",
       "  'occupation': 'Seaman',\n",
       "  'telephone': '344-219-9130',\n",
       "  'address': {'address': '1135 Hawthorne Concession road',\n",
       "   'city': 'Springfield'},\n",
       "  'credit-card': {'number': '3416 044674 28287', 'expiration-date': '06/20'}},)"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "b.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'age': 51,\n",
       "  'occupation': 'Seaman',\n",
       "  'telephone': '344-219-9130',\n",
       "  'credit-card-number': '3416 044674 28287',\n",
       "  'credit-card-expiration': '06/20',\n",
       "  'name': 'Chas Ratliff',\n",
       "  'street-address': '1135 Hawthorne Concession road',\n",
       "  'city': 'Springfield'},)"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def flatten(record):\n",
    "    return {\n",
    "        'age': record['age'],\n",
    "        'occupation': record['occupation'],\n",
    "        'telephone': record['telephone'],\n",
    "        'credit-card-number': record['credit-card']['number'],\n",
    "        'credit-card-expiration': record['credit-card']['expiration-date'],\n",
    "        'name': ' '.join(record['name']),\n",
    "        'street-address': record['address']['address'],\n",
    "        'city': record['address']['city']\n",
    "    }\n",
    "\n",
    "b.map(flatten).take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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>age</th>\n",
       "      <th>occupation</th>\n",
       "      <th>telephone</th>\n",
       "      <th>credit-card-number</th>\n",
       "      <th>credit-card-expiration</th>\n",
       "      <th>name</th>\n",
       "      <th>street-address</th>\n",
       "      <th>city</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>51</td>\n",
       "      <td>Seaman</td>\n",
       "      <td>344-219-9130</td>\n",
       "      <td>3416 044674 28287</td>\n",
       "      <td>06/20</td>\n",
       "      <td>Chas Ratliff</td>\n",
       "      <td>1135 Hawthorne Concession road</td>\n",
       "      <td>Springfield</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>53</td>\n",
       "      <td>Geophysicist</td>\n",
       "      <td>(219) 049-8427</td>\n",
       "      <td>4315 9929 2231 3721</td>\n",
       "      <td>04/22</td>\n",
       "      <td>Wilbur Cannon</td>\n",
       "      <td>871 Bernal Heights Lane</td>\n",
       "      <td>Hallandale Beach</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>24</td>\n",
       "      <td>Hospital Worker</td>\n",
       "      <td>1-899-405-6481</td>\n",
       "      <td>4494 7877 3692 6723</td>\n",
       "      <td>03/22</td>\n",
       "      <td>Spring Case</td>\n",
       "      <td>820 St. Paul Street-Calvert Hill</td>\n",
       "      <td>Chesapeake</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>60</td>\n",
       "      <td>Osteopath</td>\n",
       "      <td>1-677-679-4179</td>\n",
       "      <td>4848 0637 1092 6516</td>\n",
       "      <td>12/25</td>\n",
       "      <td>Lou Galloway</td>\n",
       "      <td>1076 Ellis Viaduct</td>\n",
       "      <td>Kennesaw</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>37</td>\n",
       "      <td>Art Historian</td>\n",
       "      <td>514-516-9013</td>\n",
       "      <td>3455 426227 43867</td>\n",
       "      <td>06/17</td>\n",
       "      <td>Thaddeus Kirk</td>\n",
       "      <td>576 Candyland Nene</td>\n",
       "      <td>Peabody</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   age       occupation       telephone   credit-card-number  \\\n",
       "0   51           Seaman    344-219-9130    3416 044674 28287   \n",
       "1   53     Geophysicist  (219) 049-8427  4315 9929 2231 3721   \n",
       "2   24  Hospital Worker  1-899-405-6481  4494 7877 3692 6723   \n",
       "3   60        Osteopath  1-677-679-4179  4848 0637 1092 6516   \n",
       "4   37    Art Historian    514-516-9013    3455 426227 43867   \n",
       "\n",
       "  credit-card-expiration           name                    street-address  \\\n",
       "0                  06/20   Chas Ratliff    1135 Hawthorne Concession road   \n",
       "1                  04/22  Wilbur Cannon           871 Bernal Heights Lane   \n",
       "2                  03/22    Spring Case  820 St. Paul Street-Calvert Hill   \n",
       "3                  12/25   Lou Galloway                1076 Ellis Viaduct   \n",
       "4                  06/17  Thaddeus Kirk                576 Candyland Nene   \n",
       "\n",
       "               city  \n",
       "0       Springfield  \n",
       "1  Hallandale Beach  \n",
       "2        Chesapeake  \n",
       "3          Kennesaw  \n",
       "4           Peabody  "
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = b.map(flatten).to_dataframe()\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Stonemason          25\n",
       "Nursing Manager     24\n",
       "Valve Technician    23\n",
       "Horticulturalist    23\n",
       "Theatrical Agent    23\n",
       "Landworker          22\n",
       "Mortician           22\n",
       "Blind Fitter        21\n",
       "Care Assistant      21\n",
       "Almoner             21\n",
       "Name: occupation, dtype: int64"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# dataframe操作\n",
    "df[df.age > 30].occupation.value_counts().nlargest(10).compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}