Base/tools/spark/learnsparkLDA/spark_pairRDD.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# pair RDD\n",
    "Spark为包含键值对类型的RDD提供了一些专有的操作，这些RDD被称为pair RDD。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "\n",
    "from pyspark import SparkConf,SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "file:///Users/fire/jupyter/data/ball2018.txt MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf = SparkConf().setMaster(\"local\").setAppName(\"My APP\")\n",
    "sc = SparkContext(conf = conf)\n",
    "\n",
    "lines = sc.textFile('file:///Users/fire/jupyter/data/ball2018.txt')\n",
    "lines"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## 1.pair RDD的操作"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('2018010',\n",
       " '2018010 2018.01.23  01081720212203  351327690   6注8085375元  71注325919元  奖池:3.9亿 详情走势')"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 使用第一个元素作为键创建一个pair RDD\n",
    "pairs = lines.map(lambda x: (x.split(\" \")[0],x))\n",
    "pairs.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[7] at RDD at PythonRDD.scala:49"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 对第二个元素进行筛选\n",
    "pairs.filter(lambda keyValue: len(keyValue[1] < 20))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[12] at RDD at PythonRDD.scala:49"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 使用reduceByKey() 和 mapValues()计算每个键对应的平均值\n",
    "\n",
    "# reduceByKey(): 合并含有相同键的值\n",
    "# mapValues(): 对pair RDD中的每个值应用一个函数而不改变键\n",
    "\n",
    "# 这里的x都是值 不是键\n",
    "pairs.mapValues(lambda x:(x,1)).reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('2018010', 1)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#### 实现单词计数\n",
    "words = lines.flatMap(lambda x: x.split(\" \"))\n",
    "#flatMap： Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.\n",
    "result = words.map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y)\n",
    "result.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[36] at RDD at PythonRDD.scala:49"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# comineByKey(createCombiner, mergeValue, mergeCombiners, partitioner)：使用不同的返回类型合并具有相同键的值\n",
    "\n",
    "#求每个键的平均值\n",
    "sumCount = words.combineByKey((lambda x:(x,1)),\n",
    "                             (lambda x, y: (x[0] + y, x[1] + 1)),\n",
    "                             (lambda x, y: (x[0] + y[0], x[1] + y[1])))\n",
    "sumCount.map(lambda key,xy: (key,xy[0]/xy[1]))\n",
    "#原书最后还有个 .collectAsMap() 但是会报错，估计和count()一样的原因，先忽略。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[47] at RDD at PythonRDD.scala:49"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 自定义reduceBy的并行度\n",
    "data = [('a',3),('b',4),('c',5)]\n",
    "sc.parallelize(data).reduceByKey(lambda x,y:x+y) #默认并行度\n",
    "sc.parallelize(data).reduceByKey(lambda x,y:x+y, 10) #自定义并行度"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'2018010 2018.01.23  01081720212203  351327690   6注8085375元  71注325919元  奖池:3.9亿 详情走势'"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 以字符串顺序对整数进行自定义排序\n",
    "\n",
    "linesort = lines.sortByKey(ascending=True, numPartitions=None, keyfunc=lambda x: str(x))\n",
    "linesort.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 额外的行动操作    暂时还是会报错\n",
    "#words.countByKey()  返对每个键对应的元素分别计数 \n",
    "#words.lookup('2018010') 返回给定键对应的所有值"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## 2.数据分区\n",
    "spark许多操作都引入了将数据根据键跨节点进行混洗的过程，所有这些操作都可以从数据分区中获益。\n",
    "\n",
    "分布式程序中通信代价很大，spark可以通过控制RDD分区方式来减少通信开销。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[55] at mapPartitions at PythonRDD.scala:129"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 自定义分区方式\n",
    "import urllib.parse\n",
    "\n",
    "def hash_domin(url):\n",
    "    return hash(urllib.parse.urlparse(url).netloc)\n",
    "\n",
    "lines.partitionBy(20, hash_domin) #创建20个分区"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "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.6.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}